mirror of
https://github.com/SigNoz/signoz.git
synced 2026-05-27 04:10:28 +01:00
Compare commits
6 Commits
host-url
...
order-pipe
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
abbb6f7137 | ||
|
|
c1f3dacc0b | ||
|
|
2067b0f634 | ||
|
|
75df0c00ef | ||
|
|
d03ff055bb | ||
|
|
2b10fbd4dd |
@@ -5,7 +5,6 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -48,14 +47,13 @@ func NewAnomalyRule(
|
||||
p *ruletypes.PostableRule,
|
||||
querier querier.Querier,
|
||||
logger *slog.Logger,
|
||||
externalURL *url.URL,
|
||||
opts ...baserules.RuleOption,
|
||||
) (*AnomalyRule, error) {
|
||||
logger.Info("creating new AnomalyRule", slog.String("rule.id", id))
|
||||
|
||||
opts = append(opts, baserules.WithLogger(logger))
|
||||
|
||||
baseRule, err := baserules.NewBaseRule(id, orgID, p, externalURL, opts...)
|
||||
baseRule, err := baserules.NewBaseRule(id, orgID, p, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ package rules
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/url"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -121,7 +120,6 @@ func TestAnomalyRule_NoData_AlertOnAbsent(t *testing.T) {
|
||||
&postableRule,
|
||||
nil,
|
||||
logger,
|
||||
mustParseURL(t, "http://localhost:8000"),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -249,8 +247,7 @@ func TestAnomalyRule_NoData_AbsentFor(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8000")
|
||||
rule, err := NewAnomalyRule("test-anomaly-rule", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL)
|
||||
rule, err := NewAnomalyRule("test-anomaly-rule", valuer.GenerateUUID(), &postableRule, nil, logger)
|
||||
require.NoError(t, err)
|
||||
|
||||
rule.provider = &mockAnomalyProvider{
|
||||
@@ -267,10 +264,3 @@ func TestAnomalyRule_NoData_AbsentFor(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func mustParseURL(t *testing.T, raw string) *url.URL {
|
||||
t.Helper()
|
||||
u, err := url.Parse(raw)
|
||||
require.NoError(t, err)
|
||||
return u
|
||||
}
|
||||
|
||||
@@ -34,7 +34,6 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
|
||||
opts.Rule,
|
||||
opts.Querier,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
baserules.WithEvalDelay(opts.ManagerOpts.EvalDelay),
|
||||
baserules.WithSQLStore(opts.SQLStore),
|
||||
baserules.WithQueryParser(opts.ManagerOpts.QueryParser),
|
||||
@@ -60,7 +59,6 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
|
||||
opts.Rule,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Prometheus,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
baserules.WithSQLStore(opts.SQLStore),
|
||||
baserules.WithQueryParser(opts.ManagerOpts.QueryParser),
|
||||
baserules.WithMetadataStore(opts.ManagerOpts.MetadataStore),
|
||||
@@ -84,7 +82,6 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
|
||||
opts.Rule,
|
||||
opts.Querier,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
baserules.WithEvalDelay(opts.ManagerOpts.EvalDelay),
|
||||
baserules.WithSQLStore(opts.SQLStore),
|
||||
baserules.WithQueryParser(opts.ManagerOpts.QueryParser),
|
||||
@@ -144,7 +141,6 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
|
||||
parsedRule,
|
||||
opts.Querier,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
baserules.WithSendAlways(),
|
||||
baserules.WithSendUnmatched(),
|
||||
baserules.WithSQLStore(opts.SQLStore),
|
||||
@@ -166,7 +162,6 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
|
||||
parsedRule,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Prometheus,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
baserules.WithSendAlways(),
|
||||
baserules.WithSendUnmatched(),
|
||||
baserules.WithSQLStore(opts.SQLStore),
|
||||
@@ -186,7 +181,6 @@ func TestNotification(opts baserules.PrepareTestRuleOptions) (int, error) {
|
||||
parsedRule,
|
||||
opts.Querier,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
baserules.WithSendAlways(),
|
||||
baserules.WithSendUnmatched(),
|
||||
baserules.WithSQLStore(opts.SQLStore),
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager"
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/alertmanagerserver"
|
||||
alertmanagermock "github.com/SigNoz/signoz/pkg/alertmanager/alertmanagertest"
|
||||
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
@@ -52,7 +51,6 @@ func TestManager_TestNotification_SendUnmatched_ThresholdRule(t *testing.T) {
|
||||
fAlert := am.(*alertmanagermock.MockAlertmanager)
|
||||
// mock set notification config
|
||||
fAlert.On("SetNotificationConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
||||
fAlert.On("Config").Return(alertmanagerserver.Config{ExternalURL: mustParseURL(t, "http://localhost:8080")})
|
||||
// for saving temp alerts that are triggered via TestNotification
|
||||
if tc.ExpectAlerts > 0 {
|
||||
fAlert.On("TestAlert", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
|
||||
@@ -168,7 +166,6 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
mockAM := am.(*alertmanagermock.MockAlertmanager)
|
||||
// mock set notification config
|
||||
mockAM.On("SetNotificationConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
||||
mockAM.On("Config").Return(alertmanagerserver.Config{ExternalURL: mustParseURL(t, "http://localhost:8080")})
|
||||
// for saving temp alerts that are triggered via TestNotification
|
||||
if tc.ExpectAlerts > 0 {
|
||||
mockAM.On("TestAlert", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
|
||||
|
||||
@@ -5,7 +5,6 @@ import (
|
||||
|
||||
amConfig "github.com/prometheus/alertmanager/config"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/alertmanagerserver"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/statsreporter"
|
||||
@@ -49,9 +48,6 @@ type Alertmanager interface {
|
||||
// DeleteChannelByID deletes a channel for the organization.
|
||||
DeleteChannelByID(context.Context, string, valuer.UUID) error
|
||||
|
||||
// Config returns the alertmanagerserver configuration.
|
||||
Config() alertmanagerserver.Config
|
||||
|
||||
// SetConfig sets the config for the organization.
|
||||
SetConfig(context.Context, *alertmanagertypes.Config) error
|
||||
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"context"
|
||||
"net/http"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/alertmanagerserver"
|
||||
"github.com/SigNoz/signoz/pkg/types/alertmanagertypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/prometheus/alertmanager/config"
|
||||
@@ -110,50 +109,6 @@ func (_c *MockAlertmanager_Collect_Call) RunAndReturn(run func(context1 context.
|
||||
return _c
|
||||
}
|
||||
|
||||
// Config provides a mock function for the type MockAlertmanager
|
||||
func (_mock *MockAlertmanager) Config() alertmanagerserver.Config {
|
||||
ret := _mock.Called()
|
||||
|
||||
if len(ret) == 0 {
|
||||
panic("no return value specified for Config")
|
||||
}
|
||||
|
||||
var r0 alertmanagerserver.Config
|
||||
if returnFunc, ok := ret.Get(0).(func() alertmanagerserver.Config); ok {
|
||||
r0 = returnFunc()
|
||||
} else {
|
||||
r0 = ret.Get(0).(alertmanagerserver.Config)
|
||||
}
|
||||
return r0
|
||||
}
|
||||
|
||||
// MockAlertmanager_Config_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Config'
|
||||
type MockAlertmanager_Config_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// Config is a helper method to define mock.On call
|
||||
func (_e *MockAlertmanager_Expecter) Config() *MockAlertmanager_Config_Call {
|
||||
return &MockAlertmanager_Config_Call{Call: _e.mock.On("Config")}
|
||||
}
|
||||
|
||||
func (_c *MockAlertmanager_Config_Call) Run(run func()) *MockAlertmanager_Config_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
run()
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockAlertmanager_Config_Call) Return(config alertmanagerserver.Config) *MockAlertmanager_Config_Call {
|
||||
_c.Call.Return(config)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockAlertmanager_Config_Call) RunAndReturn(run func() alertmanagerserver.Config) *MockAlertmanager_Config_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
|
||||
// CreateChannel provides a mock function for the type MockAlertmanager
|
||||
func (_mock *MockAlertmanager) CreateChannel(context1 context.Context, s string, v alertmanagertypes.Receiver) (*alertmanagertypes.Channel, error) {
|
||||
ret := _mock.Called(context1, s, v)
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"github.com/prometheus/common/model"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager"
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/alertmanagerserver"
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/alertmanagerstore/sqlalertmanagerstore"
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/nfmanager"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
@@ -236,10 +235,6 @@ func (provider *provider) CreateChannel(ctx context.Context, orgID string, recei
|
||||
return channel, nil
|
||||
}
|
||||
|
||||
func (provider *provider) Config() alertmanagerserver.Config {
|
||||
return provider.config.Signoz.Config
|
||||
}
|
||||
|
||||
func (provider *provider) SetConfig(ctx context.Context, config *alertmanagertypes.Config) error {
|
||||
return provider.configStore.Set(ctx, config)
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
tracesV3 "github.com/SigNoz/signoz/pkg/query-service/app/traces/v3"
|
||||
@@ -227,7 +226,7 @@ func PrepareFilters(labels map[string]string, whereClauseItems []v3.FilterItem,
|
||||
return filterItems
|
||||
}
|
||||
|
||||
func PrepareParamsForTracesV5(start, end time.Time, whereClause string) url.Values {
|
||||
func PrepareLinksToTracesV5(start, end time.Time, whereClause string) string {
|
||||
|
||||
// Traces list view expects time in nanoseconds
|
||||
tr := URLShareableTimeRange{
|
||||
@@ -239,6 +238,7 @@ func PrepareParamsForTracesV5(start, end time.Time, whereClause string) url.Valu
|
||||
options := URLShareableOptions{}
|
||||
|
||||
period, _ := json.Marshal(tr)
|
||||
urlEncodedTimeRange := url.QueryEscape(string(period))
|
||||
|
||||
linkQuery := LinkQuery{
|
||||
BuilderQuery: v3.BuilderQuery{
|
||||
@@ -265,20 +265,15 @@ func PrepareParamsForTracesV5(start, end time.Time, whereClause string) url.Valu
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(urlData)
|
||||
compositeQuery := url.QueryEscape(string(data))
|
||||
compositeQuery := url.QueryEscape(url.QueryEscape(string(data)))
|
||||
|
||||
optionsData, _ := json.Marshal(options)
|
||||
urlEncodedOptions := url.QueryEscape(string(optionsData))
|
||||
|
||||
params := url.Values{}
|
||||
params.Set("compositeQuery", compositeQuery)
|
||||
params.Set("timeRange", string(period))
|
||||
params.Set("startTime", strconv.FormatInt(tr.Start, 10))
|
||||
params.Set("endTime", strconv.FormatInt(tr.End, 10))
|
||||
params.Set("options", string(optionsData))
|
||||
return params
|
||||
return fmt.Sprintf("compositeQuery=%s&timeRange=%s&startTime=%d&endTime=%d&options=%s", compositeQuery, urlEncodedTimeRange, tr.Start, tr.End, urlEncodedOptions)
|
||||
}
|
||||
|
||||
func PrepareParamsForLogsV5(start, end time.Time, whereClause string) url.Values {
|
||||
func PrepareLinksToLogsV5(start, end time.Time, whereClause string) string {
|
||||
|
||||
// Logs list view expects time in milliseconds
|
||||
tr := URLShareableTimeRange{
|
||||
@@ -290,6 +285,7 @@ func PrepareParamsForLogsV5(start, end time.Time, whereClause string) url.Values
|
||||
options := URLShareableOptions{}
|
||||
|
||||
period, _ := json.Marshal(tr)
|
||||
urlEncodedTimeRange := url.QueryEscape(string(period))
|
||||
|
||||
linkQuery := LinkQuery{
|
||||
BuilderQuery: v3.BuilderQuery{
|
||||
@@ -316,15 +312,10 @@ func PrepareParamsForLogsV5(start, end time.Time, whereClause string) url.Values
|
||||
}
|
||||
|
||||
data, _ := json.Marshal(urlData)
|
||||
compositeQuery := url.QueryEscape(string(data))
|
||||
compositeQuery := url.QueryEscape(url.QueryEscape(string(data)))
|
||||
|
||||
optionsData, _ := json.Marshal(options)
|
||||
urlEncodedOptions := url.QueryEscape(string(optionsData))
|
||||
|
||||
params := url.Values{}
|
||||
params.Set("compositeQuery", compositeQuery)
|
||||
params.Set("timeRange", string(period))
|
||||
params.Set("startTime", strconv.FormatInt(tr.Start, 10))
|
||||
params.Set("endTime", strconv.FormatInt(tr.End, 10))
|
||||
params.Set("options", string(optionsData))
|
||||
return params
|
||||
return fmt.Sprintf("compositeQuery=%s&timeRange=%s&startTime=%d&endTime=%d&options=%s", compositeQuery, urlEncodedTimeRange, tr.Start, tr.End, urlEncodedOptions)
|
||||
}
|
||||
|
||||
@@ -958,7 +958,7 @@ func (aH *APIHandler) getRuleStateHistory(w http.ResponseWriter, r *http.Request
|
||||
|
||||
whereClause := contextlinks.PrepareFilterExpression(lbls, filterExpr, q.GroupBy)
|
||||
|
||||
res.Items[idx].RelatedLogsLink = contextlinks.PrepareParamsForLogsV5(start, end, whereClause).Encode()
|
||||
res.Items[idx].RelatedLogsLink = contextlinks.PrepareLinksToLogsV5(start, end, whereClause)
|
||||
} else if rule.AlertType == ruletypes.AlertTypeTraces {
|
||||
// TODO(srikanthccv): re-visit this and support multiple queries
|
||||
var q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]
|
||||
@@ -978,7 +978,7 @@ func (aH *APIHandler) getRuleStateHistory(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
|
||||
whereClause := contextlinks.PrepareFilterExpression(lbls, filterExpr, q.GroupBy)
|
||||
res.Items[idx].RelatedTracesLink = contextlinks.PrepareParamsForTracesV5(start, end, whereClause).Encode()
|
||||
res.Items[idx].RelatedTracesLink = contextlinks.PrepareLinksToTracesV5(start, end, whereClause)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,14 +2,11 @@ package logparsingpipeline
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"gopkg.in/yaml.v3"
|
||||
|
||||
"log/slog"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/constants"
|
||||
"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
|
||||
@@ -26,6 +23,13 @@ var (
|
||||
CodeCollectorConfigLogsPipelineNotFound = errors.MustNewCode("collector_config_logs_pipeline_not_found")
|
||||
)
|
||||
|
||||
const (
|
||||
memoryLimiterProcessor = "memory_limiter"
|
||||
memoryLimiterProcessorPrefix = "memory_limiter/"
|
||||
batchProcessor = "batch"
|
||||
batchProcessorPrefix = "batch/"
|
||||
)
|
||||
|
||||
// check if the processors already exist
|
||||
// if yes then update the processor.
|
||||
// if something doesn't exists then remove it.
|
||||
@@ -79,6 +83,14 @@ func getOtelPipelineFromConfig(config map[string]interface{}) (*otelPipeline, er
|
||||
return &p, nil
|
||||
}
|
||||
|
||||
// buildCollectorPipelineProcessorsList assembles the final processor list in the
|
||||
// required order:
|
||||
//
|
||||
// 1. memory_limiter processors (any processor named "memory_limiter" or "memory_limiter/<id>")
|
||||
// 2. other existing processors (in their original order), which may include signoz processors
|
||||
// that are not user-pipeline processors
|
||||
// 3. signoz user-pipeline processors
|
||||
// 4. batch processors (any processor named "batch" or "batch/<id>")
|
||||
func buildCollectorPipelineProcessorsList(
|
||||
currentCollectorProcessors []string,
|
||||
signozPipelineProcessorNames []string,
|
||||
@@ -86,90 +98,59 @@ func buildCollectorPipelineProcessorsList(
|
||||
lockLogsPipelineSpec.Lock()
|
||||
defer lockLogsPipelineSpec.Unlock()
|
||||
|
||||
exists := map[string]struct{}{}
|
||||
for _, v := range signozPipelineProcessorNames {
|
||||
exists[v] = struct{}{}
|
||||
// Build a set of the desired signoz processors so we can drop any stale version
|
||||
// of them (regardless of how they got into the current config) without
|
||||
// accidentally duplicating them in the output.
|
||||
desiredUserPipelineSet := make(map[string]struct{}, len(signozPipelineProcessorNames))
|
||||
for _, p := range signozPipelineProcessorNames {
|
||||
desiredUserPipelineSet[p] = struct{}{}
|
||||
}
|
||||
|
||||
// removed the old processors which are not used
|
||||
var pipeline []string
|
||||
for _, procName := range currentCollectorProcessors {
|
||||
_, isInDesiredPipelineProcs := exists[procName]
|
||||
if isInDesiredPipelineProcs || !hasSignozPipelineProcessorPrefix(procName) {
|
||||
pipeline = append(pipeline, procName)
|
||||
}
|
||||
}
|
||||
result := make([]string, 0, len(currentCollectorProcessors)+len(signozPipelineProcessorNames))
|
||||
|
||||
// create a reverse map of existing config processors and their position
|
||||
existing := map[string]int{}
|
||||
for i, p := range pipeline {
|
||||
name := p
|
||||
existing[name] = i
|
||||
}
|
||||
|
||||
// create mapping from our logsParserPipeline to position in existing processors (from current config)
|
||||
// this means, if "batch" holds position 3 in the current effective config, and 2 in our config, the map will be [2]: 3
|
||||
specVsExistingMap := map[int]int{}
|
||||
existingVsSpec := map[int]int{}
|
||||
|
||||
// go through plan and map its elements to current positions in effective config
|
||||
for i, m := range signozPipelineProcessorNames {
|
||||
if loc, ok := existing[m]; ok {
|
||||
specVsExistingMap[i] = loc
|
||||
existingVsSpec[loc] = i
|
||||
}
|
||||
}
|
||||
|
||||
lastMatched := 0
|
||||
newPipeline := []string{}
|
||||
|
||||
for i := 0; i < len(signozPipelineProcessorNames); i++ {
|
||||
m := signozPipelineProcessorNames[i]
|
||||
if loc, ok := specVsExistingMap[i]; ok {
|
||||
for j := lastMatched; j < loc; j++ {
|
||||
if hasSignozPipelineProcessorPrefix(pipeline[j]) {
|
||||
delete(specVsExistingMap, existingVsSpec[j])
|
||||
} else {
|
||||
newPipeline = append(newPipeline, pipeline[j])
|
||||
}
|
||||
// Note: logic assumes there'll be only one batch processor
|
||||
var batchProcIdx int
|
||||
var batchProcFound bool
|
||||
iteration:
|
||||
for idx, p := range currentCollectorProcessors {
|
||||
_, inDesiredSet := desiredUserPipelineSet[p]
|
||||
switch {
|
||||
// same processor exist; retain the location of pre-existing location
|
||||
case p == memoryLimiterProcessor || strings.HasPrefix(p, memoryLimiterProcessorPrefix):
|
||||
result = append(result, p)
|
||||
case hasSignozPipelineProcessorPrefix(p):
|
||||
// this processor has been dropped
|
||||
if !inDesiredSet {
|
||||
continue iteration
|
||||
} else {
|
||||
result = append(result, p)
|
||||
}
|
||||
newPipeline = append(newPipeline, pipeline[loc])
|
||||
lastMatched = loc + 1
|
||||
} else {
|
||||
newPipeline = append(newPipeline, m)
|
||||
case p == batchProcessor || strings.HasPrefix(p, batchProcessorPrefix):
|
||||
batchProcIdx = idx
|
||||
batchProcFound = true
|
||||
break iteration
|
||||
default:
|
||||
result = append(result, p)
|
||||
}
|
||||
|
||||
}
|
||||
if lastMatched < len(pipeline) {
|
||||
newPipeline = append(newPipeline, pipeline[lastMatched:]...)
|
||||
}
|
||||
|
||||
if checkDuplicateString(newPipeline) {
|
||||
// duplicates are most likely because the processor sequence in effective config conflicts
|
||||
// with the planned sequence as per planned pipeline
|
||||
return pipeline, fmt.Errorf("the effective config has an unexpected processor sequence: %v", pipeline)
|
||||
}
|
||||
|
||||
return newPipeline, nil
|
||||
}
|
||||
|
||||
func checkDuplicateString(pipeline []string) bool {
|
||||
exists := make(map[string]bool, len(pipeline))
|
||||
slog.Debug("checking duplicate processors in the pipeline", "pipeline", pipeline)
|
||||
for _, processor := range pipeline {
|
||||
name := processor
|
||||
if _, ok := exists[name]; ok {
|
||||
slog.Error(
|
||||
"duplicate processor name detected in generated collector config for log pipelines",
|
||||
"processor", processor,
|
||||
"pipeline", pipeline,
|
||||
)
|
||||
return true
|
||||
if inDesiredSet {
|
||||
// delete from desired pipeline set so they're not added twice
|
||||
delete(desiredUserPipelineSet, p)
|
||||
}
|
||||
|
||||
exists[name] = true
|
||||
}
|
||||
return false
|
||||
// add user pipelines
|
||||
for _, proc := range signozPipelineProcessorNames {
|
||||
_, add := desiredUserPipelineSet[proc]
|
||||
if add {
|
||||
result = append(result, proc)
|
||||
}
|
||||
}
|
||||
|
||||
// add batch processor and rest
|
||||
if batchProcFound {
|
||||
result = append(result, currentCollectorProcessors[batchProcIdx:]...)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
|
||||
|
||||
@@ -106,107 +106,109 @@ func TestBuildLogParsingProcessors(t *testing.T) {
|
||||
}
|
||||
|
||||
var BuildLogsPipelineTestData = []struct {
|
||||
Name string
|
||||
currentPipeline []string
|
||||
logsPipeline []string
|
||||
expectedPipeline []string
|
||||
Name string
|
||||
fromCollector []string
|
||||
userPipelines []string
|
||||
finalOutput []string
|
||||
}{
|
||||
{
|
||||
Name: "Add new pipelines",
|
||||
currentPipeline: []string{"processor1", "processor2"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b"},
|
||||
expectedPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", "processor1", "processor2"},
|
||||
Name: "Add new pipelines",
|
||||
fromCollector: []string{"processor1", "processor2"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b"},
|
||||
finalOutput: []string{"processor1", "processor2", constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b"},
|
||||
},
|
||||
{
|
||||
Name: "Add new pipeline and respect custom processors",
|
||||
currentPipeline: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", "processor2"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c"},
|
||||
expectedPipeline: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c", "processor2"},
|
||||
Name: "Add new pipeline and respect custom processors",
|
||||
fromCollector: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", "processor2"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c"},
|
||||
finalOutput: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", "processor2", constants.LogsPPLPfx + "c"},
|
||||
},
|
||||
{
|
||||
Name: "Add new pipeline and respect custom processors",
|
||||
currentPipeline: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", "processor2"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c", constants.LogsPPLPfx + "d"},
|
||||
expectedPipeline: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c", constants.LogsPPLPfx + "d", "processor2"},
|
||||
Name: "Add new pipeline and respect custom processors",
|
||||
fromCollector: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", "processor2"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c", constants.LogsPPLPfx + "d"},
|
||||
finalOutput: []string{constants.LogsPPLPfx + "a", "processor1", constants.LogsPPLPfx + "b", "processor2", constants.LogsPPLPfx + "c", constants.LogsPPLPfx + "d"},
|
||||
},
|
||||
{
|
||||
Name: "Add new pipeline and respect custom processors in the beginning and middle",
|
||||
currentPipeline: []string{"processor1", constants.LogsPPLPfx + "a", "processor2", constants.LogsPPLPfx + "b", "batch"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c"},
|
||||
expectedPipeline: []string{"processor1", constants.LogsPPLPfx + "a", "processor2", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c", "batch"},
|
||||
Name: "Add new pipeline and respect custom processors in the beginning and middle",
|
||||
fromCollector: []string{"processor1", constants.LogsPPLPfx + "a", "processor2", constants.LogsPPLPfx + "b", "batch"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c"},
|
||||
finalOutput: []string{"processor1", constants.LogsPPLPfx + "a", "processor2", constants.LogsPPLPfx + "b", constants.LogsPPLPfx + "c", "batch"},
|
||||
},
|
||||
{
|
||||
Name: "Remove old pipeline add add new",
|
||||
currentPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", "processor1", "processor2"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a"},
|
||||
expectedPipeline: []string{constants.LogsPPLPfx + "a", "processor1", "processor2"},
|
||||
Name: "Remove old pipeline add add new",
|
||||
fromCollector: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "b", "processor1", "processor2"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a"},
|
||||
finalOutput: []string{constants.LogsPPLPfx + "a", "processor1", "processor2"},
|
||||
},
|
||||
{
|
||||
Name: "Remove old pipeline from middle",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "a", "processor3", constants.LogsPPLPfx + "b", "batch"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a"},
|
||||
expectedPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "a", "processor3", "batch"},
|
||||
Name: "Remove old pipeline from middle",
|
||||
fromCollector: []string{"processor1", "processor2", constants.LogsPPLPfx + "a", "processor3", constants.LogsPPLPfx + "b", "batch"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a"},
|
||||
finalOutput: []string{"processor1", "processor2", constants.LogsPPLPfx + "a", "processor3", "batch"},
|
||||
},
|
||||
{
|
||||
Name: "Remove old pipeline from middle and add new pipeline",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "a", "processor3", constants.LogsPPLPfx + "b", "batch"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "c"},
|
||||
expectedPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "c", "processor3", "batch"},
|
||||
Name: "Remove old pipeline from middle and add new pipeline",
|
||||
fromCollector: []string{"memory_limiter", "processor1", "processor2", constants.LogsPPLPfx + "a", "processor3", constants.LogsPPLPfx + "b", "batch"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "c"},
|
||||
finalOutput: []string{"memory_limiter", "processor1", "processor2", constants.LogsPPLPfx + "a", "processor3", constants.LogsPPLPfx + "c", "batch"},
|
||||
},
|
||||
{
|
||||
Name: "Remove multiple old pipelines from middle and add multiple new ones",
|
||||
currentPipeline: []string{"processor1", constants.LogsPPLPfx + "a", "processor2", constants.LogsPPLPfx + "b", "processor3", constants.LogsPPLPfx + "c", "processor4", constants.LogsPPLPfx + "d", "processor5", "batch"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "a1", constants.LogsPPLPfx + "c", constants.LogsPPLPfx + "c1"},
|
||||
expectedPipeline: []string{"processor1", constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "a1", "processor2", "processor3", constants.LogsPPLPfx + "c", constants.LogsPPLPfx + "c1", "processor4", "processor5", "batch"},
|
||||
},
|
||||
|
||||
// working
|
||||
{
|
||||
Name: "rearrange pipelines",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a"},
|
||||
expectedPipeline: []string{"processor1", "processor2", "processor3", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", "batch"},
|
||||
Name: "Remove multiple old pipelines from middle and add multiple new ones",
|
||||
fromCollector: []string{"processor1", constants.LogsPPLPfx + "a", "processor2", constants.LogsPPLPfx + "b", "processor3", constants.LogsPPLPfx + "c", "processor4", constants.LogsPPLPfx + "d", "processor5", "batch"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "a", constants.LogsPPLPfx + "a1", constants.LogsPPLPfx + "c", constants.LogsPPLPfx + "c1"},
|
||||
finalOutput: []string{"processor1", constants.LogsPPLPfx + "a", "processor2", "processor3", constants.LogsPPLPfx + "c", "processor4", "processor5", constants.LogsPPLPfx + "a1", constants.LogsPPLPfx + "c1", "batch"},
|
||||
},
|
||||
{
|
||||
Name: "rearrange pipelines with new processor",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_c"},
|
||||
expectedPipeline: []string{"processor1", "processor2", "processor3", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_c", "batch"},
|
||||
// expectedPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "_b", "processor3", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_c", "batch"},
|
||||
Name: "rearrange pipelines",
|
||||
fromCollector: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a"},
|
||||
finalOutput: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch"},
|
||||
},
|
||||
{
|
||||
Name: "delete processor",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch"},
|
||||
logsPipeline: []string{},
|
||||
expectedPipeline: []string{"processor1", "processor2", "processor3", "batch"},
|
||||
Name: "rearrange pipelines with new processor",
|
||||
fromCollector: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_c"},
|
||||
finalOutput: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_c", "batch"},
|
||||
},
|
||||
{
|
||||
Name: "last to first",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", "processor4", constants.LogsPPLPfx + "_b", "batch", constants.LogsPPLPfx + "_c"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_b"},
|
||||
expectedPipeline: []string{"processor1", "processor2", "processor3", "processor4", "batch", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_b"},
|
||||
Name: "delete processor",
|
||||
fromCollector: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch"},
|
||||
userPipelines: []string{},
|
||||
finalOutput: []string{"processor1", "processor2", "processor3", "batch"},
|
||||
},
|
||||
{
|
||||
Name: "multiple rearrange pipelines",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch", constants.LogsPPLPfx + "_c", "processor4", "processor5", constants.LogsPPLPfx + "_d", "processor6", "processor7"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e"},
|
||||
expectedPipeline: []string{"processor1", "processor2", "processor3", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", "batch", "processor4", "processor5", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e", "processor6", "processor7"},
|
||||
Name: "last to first",
|
||||
fromCollector: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", "processor4", constants.LogsPPLPfx + "_b", "batch", constants.LogsPPLPfx + "_c"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_b"},
|
||||
finalOutput: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", "processor4", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_c", "batch", constants.LogsPPLPfx + "_c"},
|
||||
},
|
||||
{
|
||||
Name: "multiple rearrange with new pipelines",
|
||||
currentPipeline: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch", constants.LogsPPLPfx + "_c", "processor4", "processor5", constants.LogsPPLPfx + "_d", "processor6", "processor7"},
|
||||
logsPipeline: []string{constants.LogsPPLPfx + "_z", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e"},
|
||||
expectedPipeline: []string{constants.LogsPPLPfx + "_z", "processor1", "processor2", "processor3", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", "batch", "processor4", "processor5", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e", "processor6", "processor7"},
|
||||
Name: "multiple rearrange pipelines",
|
||||
fromCollector: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch", constants.LogsPPLPfx + "_c", "processor4", "processor5", constants.LogsPPLPfx + "_d", "processor6", "processor7"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e"},
|
||||
finalOutput: []string{"processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e", "batch", constants.LogsPPLPfx + "_c", "processor4", "processor5", constants.LogsPPLPfx + "_d", "processor6", "processor7"},
|
||||
},
|
||||
{
|
||||
Name: "multiple rearrange with new pipelines",
|
||||
fromCollector: []string{"memory_limiter", "processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", "batch", constants.LogsPPLPfx + "_c", "processor4", "processor5", constants.LogsPPLPfx + "_d", "processor6", "processor7"},
|
||||
userPipelines: []string{constants.LogsPPLPfx + "_z", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_a", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e"},
|
||||
finalOutput: []string{"memory_limiter", "processor1", "processor2", constants.LogsPPLPfx + "_a", "processor3", constants.LogsPPLPfx + "_b", constants.LogsPPLPfx + "_z", constants.LogsPPLPfx + "_d", constants.LogsPPLPfx + "_c", constants.LogsPPLPfx + "_e", "batch", constants.LogsPPLPfx + "_c", "processor4", "processor5", constants.LogsPPLPfx + "_d", "processor6", "processor7"},
|
||||
},
|
||||
{
|
||||
Name: "Prefixed proc in desired set not duplicated from others",
|
||||
fromCollector: []string{"memory_limiter/logs", "custom_proc", "resourcedetection", "batch/logs"},
|
||||
userPipelines: []string{"custom_proc", constants.LogsPPLPfx + "a"},
|
||||
finalOutput: []string{"memory_limiter/logs", "custom_proc", "resourcedetection", constants.LogsPPLPfx + "a", "batch/logs"},
|
||||
},
|
||||
}
|
||||
|
||||
func TestBuildLogsPipeline(t *testing.T) {
|
||||
for _, test := range BuildLogsPipelineTestData {
|
||||
Convey(test.Name, t, func() {
|
||||
v, err := buildCollectorPipelineProcessorsList(test.currentPipeline, test.logsPipeline)
|
||||
v, err := buildCollectorPipelineProcessorsList(test.fromCollector, test.userPipelines)
|
||||
So(err, ShouldBeNil)
|
||||
fmt.Println(test.Name, "\n", test.currentPipeline, "\n", v, "\n", test.expectedPipeline)
|
||||
So(v, ShouldResemble, test.expectedPipeline)
|
||||
So(v, ShouldResemble, test.finalOutput)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -24,7 +23,7 @@ type BaseRule struct {
|
||||
id string
|
||||
name string
|
||||
orgID valuer.UUID
|
||||
externalURL *url.URL
|
||||
source string
|
||||
handledRestart bool
|
||||
|
||||
// Type of the rule
|
||||
@@ -139,13 +138,7 @@ func WithRuleStateHistoryModule(module rulestatehistory.Module) RuleOption {
|
||||
}
|
||||
}
|
||||
|
||||
func NewBaseRule(
|
||||
id string,
|
||||
orgID valuer.UUID,
|
||||
p *ruletypes.PostableRule,
|
||||
externalURL *url.URL,
|
||||
opts ...RuleOption,
|
||||
) (*BaseRule, error) {
|
||||
func NewBaseRule(id string, orgID valuer.UUID, p *ruletypes.PostableRule, opts ...RuleOption) (*BaseRule, error) {
|
||||
threshold, err := p.RuleCondition.Thresholds.GetRuleThreshold()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -158,8 +151,8 @@ func NewBaseRule(
|
||||
baseRule := &BaseRule{
|
||||
id: id,
|
||||
orgID: orgID,
|
||||
externalURL: externalURL,
|
||||
name: p.AlertName,
|
||||
source: p.Source,
|
||||
typ: p.AlertType,
|
||||
ruleCondition: p.RuleCondition,
|
||||
evalWindow: p.EvalWindow,
|
||||
@@ -248,17 +241,7 @@ func (r *BaseRule) Annotations() ruletypes.Labels { return r.annotations }
|
||||
func (r *BaseRule) PreferredChannels() []string { return r.preferredChannels }
|
||||
|
||||
func (r *BaseRule) GeneratorURL() string {
|
||||
params := url.Values{}
|
||||
params.Set("ruleId", r.id)
|
||||
return r.ExternalURL("alerts/overview", params)
|
||||
}
|
||||
|
||||
func (r *BaseRule) ExternalURL(path string, params url.Values) string {
|
||||
u := r.externalURL.JoinPath(path)
|
||||
if len(params) > 0 {
|
||||
u.RawQuery = params.Encode()
|
||||
}
|
||||
return u.String()
|
||||
return ruletypes.PrepareRuleGeneratorURL(r.ID(), r.source)
|
||||
}
|
||||
|
||||
func (r *BaseRule) SelectedQuery(ctx context.Context) string {
|
||||
|
||||
@@ -3,7 +3,6 @@ package rules
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -19,13 +18,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
func mustParseURL(t *testing.T, raw string) *url.URL {
|
||||
t.Helper()
|
||||
u, err := url.Parse(raw)
|
||||
require.NoError(t, err)
|
||||
return u
|
||||
}
|
||||
|
||||
// createTestSeries creates a *qbtypes.TimeSeries with the given labels and optional points
|
||||
// so we don't exactly need the points in the series because the labels are used to determine if the series is new or old
|
||||
// we use the labels to create a lookup key for the series and then check the first_seen timestamp for the series in the metadata table
|
||||
@@ -689,15 +681,7 @@ func TestBaseRule_FilterNewSeries(t *testing.T) {
|
||||
}
|
||||
|
||||
// Create BaseRule using NewBaseRule
|
||||
rule, err := NewBaseRule(
|
||||
"test-rule",
|
||||
valuer.GenerateUUID(),
|
||||
&postableRule,
|
||||
mustParseURL(t, "http://localhost:8080"),
|
||||
WithQueryParser(queryParser),
|
||||
WithLogger(logger),
|
||||
WithMetadataStore(mockMetadataStore),
|
||||
)
|
||||
rule, err := NewBaseRule("test-rule", valuer.GenerateUUID(), &postableRule, WithQueryParser(queryParser), WithLogger(logger), WithMetadataStore(mockMetadataStore))
|
||||
require.NoError(t, err)
|
||||
|
||||
filteredSeries, err := rule.FilterNewSeries(context.Background(), tt.evalTime, tt.series)
|
||||
@@ -739,69 +723,6 @@ func TestBaseRule_FilterNewSeries(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBaseRule_ExternalURL(t *testing.T) {
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
externalURL *url.URL
|
||||
want string
|
||||
}{
|
||||
{name: "default value returned as-is", externalURL: mustParseURL(t, "http://localhost:8080"), want: "http://localhost:8080"},
|
||||
{name: "configured https host", externalURL: mustParseURL(t, "https://signoz.example.com"), want: "https://signoz.example.com"},
|
||||
{name: "configured host with port", externalURL: mustParseURL(t, "http://signoz.internal:3301"), want: "http://signoz.internal:3301"},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
p := createPostableRule(&ruletypes.AlertCompositeQuery{})
|
||||
r, err := NewBaseRule("some-id", valuer.GenerateUUID(), &p, tc.externalURL)
|
||||
if err != nil {
|
||||
t.Errorf("unexpected error: %v", err)
|
||||
}
|
||||
require.Equal(t, tc.want, r.ExternalURL("", nil))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBaseRule_GeneratorURL(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
ruleID string
|
||||
externalURL *url.URL
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "configured external URL",
|
||||
ruleID: "abc",
|
||||
externalURL: mustParseURL(t, "https://signoz.example.com"),
|
||||
want: "https://signoz.example.com/alerts/overview?ruleId=abc",
|
||||
},
|
||||
{
|
||||
name: "default external URL is used as-is",
|
||||
ruleID: "abc",
|
||||
externalURL: mustParseURL(t, "http://localhost:8080"),
|
||||
want: "http://localhost:8080/alerts/overview?ruleId=abc",
|
||||
},
|
||||
{
|
||||
name: "external URL with base path is preserved",
|
||||
ruleID: "abc",
|
||||
externalURL: mustParseURL(t, "https://signoz.example.com/signoz"),
|
||||
want: "https://signoz.example.com/signoz/alerts/overview?ruleId=abc",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
p := createPostableRule(&ruletypes.AlertCompositeQuery{})
|
||||
r, err := NewBaseRule(tc.ruleID, valuer.GenerateUUID(), &p, tc.externalURL)
|
||||
if err != nil {
|
||||
t.Errorf("unexpected error: %v", err)
|
||||
}
|
||||
require.Equal(t, tc.want, r.GeneratorURL())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// labelsKey creates a deterministic string key from a labels map
|
||||
// This is used to group series by their unique label combinations
|
||||
func labelsKey(lbls []*qbtypes.Label) string {
|
||||
|
||||
@@ -150,7 +150,6 @@ func defaultPrepareTaskFunc(opts PrepareTaskOptions) (Task, error) {
|
||||
opts.Rule,
|
||||
opts.Querier,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
WithEvalDelay(opts.ManagerOpts.EvalDelay),
|
||||
WithSQLStore(opts.SQLStore),
|
||||
WithQueryParser(opts.ManagerOpts.QueryParser),
|
||||
@@ -175,7 +174,6 @@ func defaultPrepareTaskFunc(opts PrepareTaskOptions) (Task, error) {
|
||||
opts.Rule,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Prometheus,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
WithSQLStore(opts.SQLStore),
|
||||
WithQueryParser(opts.ManagerOpts.QueryParser),
|
||||
WithMetadataStore(opts.ManagerOpts.MetadataStore),
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager"
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/alertmanagerserver"
|
||||
alertmanagermock "github.com/SigNoz/signoz/pkg/alertmanager/alertmanagertest"
|
||||
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
|
||||
"github.com/SigNoz/signoz/pkg/prometheus"
|
||||
@@ -51,7 +50,6 @@ func TestManager_TestNotification_SendUnmatched_ThresholdRule(t *testing.T) {
|
||||
mockAM := am.(*alertmanagermock.MockAlertmanager)
|
||||
// mock set notification config
|
||||
mockAM.On("SetNotificationConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
||||
mockAM.On("Config").Return(alertmanagerserver.Config{ExternalURL: mustParseURL(t, "http://localhost:8080")})
|
||||
// for saving temp alerts that are triggered via TestNotification
|
||||
if tc.ExpectAlerts > 0 {
|
||||
mockAM.On("TestAlert", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
|
||||
@@ -164,7 +162,6 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
|
||||
mockAM := am.(*alertmanagermock.MockAlertmanager)
|
||||
// mock set notification config
|
||||
mockAM.On("SetNotificationConfig", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
||||
mockAM.On("Config").Return(alertmanagerserver.Config{ExternalURL: mustParseURL(t, "http://localhost:8080")})
|
||||
// for saving temp alerts that are triggered via TestNotification
|
||||
if tc.ExpectAlerts > 0 {
|
||||
mockAM.On("TestAlert", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
|
||||
|
||||
@@ -5,7 +5,6 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/prometheus/model/labels"
|
||||
@@ -35,12 +34,11 @@ func NewPromRule(
|
||||
postableRule *ruletypes.PostableRule,
|
||||
logger *slog.Logger,
|
||||
prometheus prometheus.Prometheus,
|
||||
externalURL *url.URL,
|
||||
opts ...RuleOption,
|
||||
) (*PromRule, error) {
|
||||
opts = append(opts, WithLogger(logger))
|
||||
|
||||
baseRule, err := NewBaseRule(id, orgID, postableRule, externalURL, opts...)
|
||||
baseRule, err := NewBaseRule(id, orgID, postableRule, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -704,8 +704,7 @@ func TestPromRuleEval(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, nil, externalUrl)
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, nil)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
@@ -968,8 +967,7 @@ func TestPromRuleUnitCombinations(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
@@ -1085,8 +1083,7 @@ func TestPromRuleNoData(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
@@ -1319,8 +1316,7 @@ func TestMultipleThresholdPromRule(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
rule, err := NewPromRule("69", valuer.GenerateUUID(), &postableRule, logger, promProvider)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
promProvider.Close()
|
||||
@@ -1457,8 +1453,7 @@ func TestPromRule_NoData(t *testing.T) {
|
||||
_ = promProvider.Close()
|
||||
}()
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewPromRule("some-id", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
rule, err := NewPromRule("some-id", valuer.GenerateUUID(), &postableRule, logger, promProvider)
|
||||
require.NoError(t, err)
|
||||
|
||||
alertsFound, err := rule.Eval(context.Background(), evalTime)
|
||||
@@ -1608,8 +1603,7 @@ func TestPromRule_NoData_AbsentFor(t *testing.T) {
|
||||
_ = promProvider.Close()
|
||||
}()
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewPromRule("some-id", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
rule, err := NewPromRule("some-id", valuer.GenerateUUID(), &postableRule, logger, promProvider)
|
||||
require.NoError(t, err)
|
||||
|
||||
// First eval with data - should NOT alert, but populates lastTimestampWithDatapoints
|
||||
@@ -1768,8 +1762,7 @@ func TestPromRuleEval_RequireMinPoints(t *testing.T) {
|
||||
_ = promProvider.Close()
|
||||
}()
|
||||
|
||||
externalUrl := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewPromRule("some-id", valuer.GenerateUUID(), &postableRule, logger, promProvider, externalUrl)
|
||||
rule, err := NewPromRule("some-id", valuer.GenerateUUID(), &postableRule, logger, promProvider)
|
||||
require.NoError(t, err)
|
||||
|
||||
alertsFound, err := rule.Eval(context.Background(), evalTime)
|
||||
|
||||
@@ -49,7 +49,6 @@ func defaultTestNotification(opts PrepareTestRuleOptions) (int, error) {
|
||||
parsedRule,
|
||||
opts.Querier,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
WithSendAlways(),
|
||||
WithSendUnmatched(),
|
||||
WithSQLStore(opts.SQLStore),
|
||||
@@ -71,7 +70,6 @@ func defaultTestNotification(opts PrepareTestRuleOptions) (int, error) {
|
||||
parsedRule,
|
||||
opts.Logger,
|
||||
opts.ManagerOpts.Prometheus,
|
||||
opts.ManagerOpts.Alertmanager.Config().ExternalURL,
|
||||
WithSendAlways(),
|
||||
WithSendUnmatched(),
|
||||
WithSQLStore(opts.SQLStore),
|
||||
|
||||
@@ -38,14 +38,13 @@ func NewThresholdRule(
|
||||
p *ruletypes.PostableRule,
|
||||
querier querier.Querier,
|
||||
logger *slog.Logger,
|
||||
externalURL *url.URL,
|
||||
opts ...RuleOption,
|
||||
) (*ThresholdRule, error) {
|
||||
logger.Info("creating new ThresholdRule", slog.String("rule.id", id))
|
||||
|
||||
opts = append(opts, WithLogger(logger))
|
||||
|
||||
baseRule, err := NewBaseRule(id, orgID, p, externalURL, opts...)
|
||||
baseRule, err := NewBaseRule(id, orgID, p, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -56,6 +55,17 @@ func NewThresholdRule(
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (r *ThresholdRule) hostFromSource() string {
|
||||
parsedURL, err := url.Parse(r.source)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
if parsedURL.Port() != "" {
|
||||
return fmt.Sprintf("%s://%s:%s", parsedURL.Scheme, parsedURL.Hostname(), parsedURL.Port())
|
||||
}
|
||||
return fmt.Sprintf("%s://%s", parsedURL.Scheme, parsedURL.Hostname())
|
||||
}
|
||||
|
||||
func (r *ThresholdRule) Type() ruletypes.RuleType {
|
||||
return ruletypes.RuleTypeThreshold
|
||||
}
|
||||
@@ -85,19 +95,19 @@ func (r *ThresholdRule) prepareQueryRange(ctx context.Context, ts time.Time) (*q
|
||||
return req, nil
|
||||
}
|
||||
|
||||
func (r *ThresholdRule) prepareParamsForLogs(ctx context.Context, ts time.Time, lbls ruletypes.Labels) url.Values {
|
||||
func (r *ThresholdRule) prepareLinksToLogs(ctx context.Context, ts time.Time, lbls ruletypes.Labels) string {
|
||||
selectedQuery := r.SelectedQuery(ctx)
|
||||
|
||||
qr, err := r.prepareQueryRange(ctx, ts)
|
||||
if err != nil {
|
||||
return nil
|
||||
return ""
|
||||
}
|
||||
start := time.UnixMilli(int64(qr.Start))
|
||||
end := time.UnixMilli(int64(qr.End))
|
||||
|
||||
// TODO(srikanthccv): handle formula queries
|
||||
if selectedQuery < "A" || selectedQuery > "Z" {
|
||||
return nil
|
||||
return ""
|
||||
}
|
||||
|
||||
var q qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
||||
@@ -112,7 +122,7 @@ func (r *ThresholdRule) prepareParamsForLogs(ctx context.Context, ts time.Time,
|
||||
}
|
||||
|
||||
if q.Signal != telemetrytypes.SignalLogs {
|
||||
return nil
|
||||
return ""
|
||||
}
|
||||
|
||||
filterExpr := ""
|
||||
@@ -122,22 +132,22 @@ func (r *ThresholdRule) prepareParamsForLogs(ctx context.Context, ts time.Time,
|
||||
|
||||
whereClause := contextlinks.PrepareFilterExpression(lbls.Map(), filterExpr, q.GroupBy)
|
||||
|
||||
return contextlinks.PrepareParamsForLogsV5(start, end, whereClause)
|
||||
return contextlinks.PrepareLinksToLogsV5(start, end, whereClause)
|
||||
}
|
||||
|
||||
func (r *ThresholdRule) prepareParamsForTraces(ctx context.Context, ts time.Time, lbls ruletypes.Labels) url.Values {
|
||||
func (r *ThresholdRule) prepareLinksToTraces(ctx context.Context, ts time.Time, lbls ruletypes.Labels) string {
|
||||
selectedQuery := r.SelectedQuery(ctx)
|
||||
|
||||
qr, err := r.prepareQueryRange(ctx, ts)
|
||||
if err != nil {
|
||||
return nil
|
||||
return ""
|
||||
}
|
||||
start := time.UnixMilli(int64(qr.Start))
|
||||
end := time.UnixMilli(int64(qr.End))
|
||||
|
||||
// TODO(srikanthccv): handle formula queries
|
||||
if selectedQuery < "A" || selectedQuery > "Z" {
|
||||
return nil
|
||||
return ""
|
||||
}
|
||||
|
||||
var q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]
|
||||
@@ -152,7 +162,7 @@ func (r *ThresholdRule) prepareParamsForTraces(ctx context.Context, ts time.Time
|
||||
}
|
||||
|
||||
if q.Signal != telemetrytypes.SignalTraces {
|
||||
return nil
|
||||
return ""
|
||||
}
|
||||
|
||||
filterExpr := ""
|
||||
@@ -162,7 +172,7 @@ func (r *ThresholdRule) prepareParamsForTraces(ctx context.Context, ts time.Time
|
||||
|
||||
whereClause := contextlinks.PrepareFilterExpression(lbls.Map(), filterExpr, q.GroupBy)
|
||||
|
||||
return contextlinks.PrepareParamsForTracesV5(start, end, whereClause)
|
||||
return contextlinks.PrepareLinksToTracesV5(start, end, whereClause)
|
||||
}
|
||||
|
||||
func (r *ThresholdRule) buildAndRunQuery(ctx context.Context, orgID valuer.UUID, ts time.Time) (ruletypes.Vector, error) {
|
||||
@@ -339,18 +349,16 @@ func (r *ThresholdRule) Eval(ctx context.Context, ts time.Time) (int, error) {
|
||||
// label set, but different timestamps, together.
|
||||
switch r.typ {
|
||||
case ruletypes.AlertTypeTraces:
|
||||
params := r.prepareParamsForTraces(ctx, ts, smpl.Metric)
|
||||
if len(params) > 0 {
|
||||
link := r.ExternalURL("traces-explorer", params)
|
||||
r.logger.InfoContext(ctx, "adding traces link to annotations", slog.String("annotation.link", link))
|
||||
annotations = append(annotations, ruletypes.Label{Name: ruletypes.AnnotationRelatedTraces, Value: link})
|
||||
link := r.prepareLinksToTraces(ctx, ts, smpl.Metric)
|
||||
if link != "" && r.hostFromSource() != "" {
|
||||
r.logger.InfoContext(ctx, "adding traces link to annotations", slog.String("annotation.link", fmt.Sprintf("%s/traces-explorer?%s", r.hostFromSource(), link)))
|
||||
annotations = append(annotations, ruletypes.Label{Name: ruletypes.AnnotationRelatedTraces, Value: fmt.Sprintf("%s/traces-explorer?%s", r.hostFromSource(), link)})
|
||||
}
|
||||
case ruletypes.AlertTypeLogs:
|
||||
params := r.prepareParamsForLogs(ctx, ts, smpl.Metric)
|
||||
if len(params) > 0 {
|
||||
link := r.ExternalURL("logs/logs-explorer", params)
|
||||
r.logger.InfoContext(ctx, "adding logs link to annotations", slog.String("annotation.link", link))
|
||||
annotations = append(annotations, ruletypes.Label{Name: ruletypes.AnnotationRelatedLogs, Value: link})
|
||||
link := r.prepareLinksToLogs(ctx, ts, smpl.Metric)
|
||||
if link != "" && r.hostFromSource() != "" {
|
||||
r.logger.InfoContext(ctx, "adding logs link to annotations", slog.String("annotation.link", fmt.Sprintf("%s/logs/logs-explorer?%s", r.hostFromSource(), link)))
|
||||
annotations = append(annotations, ruletypes.Label{Name: ruletypes.AnnotationRelatedLogs, Value: fmt.Sprintf("%s/logs/logs-explorer?%s", r.hostFromSource(), link)})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -69,8 +69,7 @@ func TestThresholdRuleEvalWithoutRecoveryTarget(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
assert.NoError(t, err)
|
||||
|
||||
values := c.values
|
||||
@@ -142,7 +141,7 @@ func TestNormalizeLabelName(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareParamsForLogs(t *testing.T) {
|
||||
func TestPrepareLinksToLogs(t *testing.T) {
|
||||
postableRule := ruletypes.PostableRule{
|
||||
AlertName: "Tricky Condition Tests",
|
||||
AlertType: ruletypes.AlertTypeLogs,
|
||||
@@ -188,20 +187,16 @@ func TestPrepareParamsForLogs(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
assert.NoError(t, err)
|
||||
|
||||
ts := time.UnixMilli(1705469040000)
|
||||
|
||||
params := rule.prepareParamsForLogs(context.Background(), ts, ruletypes.Labels{}).Encode()
|
||||
assert.Contains(t, params, "&timeRange=%7B%22start%22%3A1705468620000%2C%22end%22%3A1705468920000%2C%22pageSize%22%3A100%7D")
|
||||
assert.Contains(t, params, "&startTime=1705468620000")
|
||||
assert.Contains(t, params, "&endTime=1705468920000")
|
||||
link := rule.prepareLinksToLogs(context.Background(), ts, ruletypes.Labels{})
|
||||
assert.Contains(t, link, "&timeRange=%7B%22start%22%3A1705468620000%2C%22end%22%3A1705468920000%2C%22pageSize%22%3A100%7D&startTime=1705468620000&endTime=1705468920000")
|
||||
}
|
||||
|
||||
func TestPrepareParamsForLogsFilterExpression(t *testing.T) {
|
||||
func TestPrepareLinksToLogsFilterExpression(t *testing.T) {
|
||||
postableRule := ruletypes.PostableRule{
|
||||
AlertName: "Tricky Condition Tests",
|
||||
AlertType: ruletypes.AlertTypeLogs,
|
||||
@@ -251,17 +246,16 @@ func TestPrepareParamsForLogsFilterExpression(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
externalURL := mustParseURL(t, "http://localhost:8000")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
assert.NoError(t, err)
|
||||
|
||||
ts := time.UnixMilli(1753527163000)
|
||||
|
||||
params := rule.prepareParamsForLogs(context.Background(), ts, ruletypes.Labels{}).Encode()
|
||||
assert.Contains(t, params, "compositeQuery=%257B%2522queryType%2522%253A%2522builder%2522%252C%2522builder%2522%253A%257B%2522queryData%2522%253A%255B%257B%2522queryName%2522%253A%2522A%2522%252C%2522stepInterval%2522%253A60%252C%2522dataSource%2522%253A%2522logs%2522%252C%2522aggregateOperator%2522%253A%2522noop%2522%252C%2522aggregateAttribute%2522%253A%257B%2522key%2522%253A%2522%2522%252C%2522dataType%2522%253A%2522%2522%252C%2522type%2522%253A%2522%2522%252C%2522isColumn%2522%253Afalse%252C%2522isJSON%2522%253Afalse%257D%252C%2522expression%2522%253A%2522A%2522%252C%2522disabled%2522%253Afalse%252C%2522limit%2522%253A0%252C%2522offset%2522%253A0%252C%2522pageSize%2522%253A0%252C%2522ShiftBy%2522%253A0%252C%2522IsAnomaly%2522%253Afalse%252C%2522QueriesUsedInFormula%2522%253Anull%252C%2522filter%2522%253A%257B%2522expression%2522%253A%2522service.name%2BEXISTS%2522%257D%257D%255D%252C%2522queryFormulas%2522%253A%255B%255D%257D%257D&endTime=1753527000000&options=%7B%22maxLines%22%3A0%2C%22format%22%3A%22%22%2C%22selectColumns%22%3Anull%7D&startTime=1753526700000&timeRange=%7B%22start%22%3A1753526700000%2C%22end%22%3A1753527000000%2C%22pageSize%22%3A100%7D")
|
||||
link := rule.prepareLinksToLogs(context.Background(), ts, ruletypes.Labels{})
|
||||
assert.Contains(t, link, "compositeQuery=%257B%2522queryType%2522%253A%2522builder%2522%252C%2522builder%2522%253A%257B%2522queryData%2522%253A%255B%257B%2522queryName%2522%253A%2522A%2522%252C%2522stepInterval%2522%253A60%252C%2522dataSource%2522%253A%2522logs%2522%252C%2522aggregateOperator%2522%253A%2522noop%2522%252C%2522aggregateAttribute%2522%253A%257B%2522key%2522%253A%2522%2522%252C%2522dataType%2522%253A%2522%2522%252C%2522type%2522%253A%2522%2522%252C%2522isColumn%2522%253Afalse%252C%2522isJSON%2522%253Afalse%257D%252C%2522expression%2522%253A%2522A%2522%252C%2522disabled%2522%253Afalse%252C%2522limit%2522%253A0%252C%2522offset%2522%253A0%252C%2522pageSize%2522%253A0%252C%2522ShiftBy%2522%253A0%252C%2522IsAnomaly%2522%253Afalse%252C%2522QueriesUsedInFormula%2522%253Anull%252C%2522filter%2522%253A%257B%2522expression%2522%253A%2522service.name%2BEXISTS%2522%257D%257D%255D%252C%2522queryFormulas%2522%253A%255B%255D%257D%257D&timeRange=%7B%22start%22%3A1753526700000%2C%22end%22%3A1753527000000%2C%22pageSize%22%3A100%7D&startTime=1753526700000&endTime=1753527000000&options=%7B%22maxLines%22%3A0%2C%22format%22%3A%22%22%2C%22selectColumns%22%3Anull%7D")
|
||||
}
|
||||
|
||||
func TestPrepareParamsForTracesFilterExpression(t *testing.T) {
|
||||
func TestPrepareLinksToTracesFilterExpression(t *testing.T) {
|
||||
postableRule := ruletypes.PostableRule{
|
||||
AlertName: "Tricky Condition Tests",
|
||||
AlertType: ruletypes.AlertTypeTraces,
|
||||
@@ -311,17 +305,16 @@ func TestPrepareParamsForTracesFilterExpression(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
externalURL := mustParseURL(t, "http://localhost:8000")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
assert.NoError(t, err)
|
||||
|
||||
ts := time.UnixMilli(1753527163000)
|
||||
|
||||
params := rule.prepareParamsForTraces(context.Background(), ts, ruletypes.Labels{}).Encode()
|
||||
assert.Contains(t, params, "compositeQuery=%257B%2522queryType%2522%253A%2522builder%2522%252C%2522builder%2522%253A%257B%2522queryData%2522%253A%255B%257B%2522queryName%2522%253A%2522A%2522%252C%2522stepInterval%2522%253A60%252C%2522dataSource%2522%253A%2522traces%2522%252C%2522aggregateOperator%2522%253A%2522noop%2522%252C%2522aggregateAttribute%2522%253A%257B%2522key%2522%253A%2522%2522%252C%2522dataType%2522%253A%2522%2522%252C%2522type%2522%253A%2522%2522%252C%2522isColumn%2522%253Afalse%252C%2522isJSON%2522%253Afalse%257D%252C%2522expression%2522%253A%2522A%2522%252C%2522disabled%2522%253Afalse%252C%2522limit%2522%253A0%252C%2522offset%2522%253A0%252C%2522pageSize%2522%253A0%252C%2522ShiftBy%2522%253A0%252C%2522IsAnomaly%2522%253Afalse%252C%2522QueriesUsedInFormula%2522%253Anull%252C%2522filter%2522%253A%257B%2522expression%2522%253A%2522service.name%2BEXISTS%2522%257D%257D%255D%252C%2522queryFormulas%2522%253A%255B%255D%257D%257D&endTime=1753527000000000000&options=%7B%22maxLines%22%3A0%2C%22format%22%3A%22%22%2C%22selectColumns%22%3Anull%7D&startTime=1753526700000000000&timeRange=%7B%22start%22%3A1753526700000000000%2C%22end%22%3A1753527000000000000%2C%22pageSize%22%3A100%7D")
|
||||
link := rule.prepareLinksToTraces(context.Background(), ts, ruletypes.Labels{})
|
||||
assert.Contains(t, link, "compositeQuery=%257B%2522queryType%2522%253A%2522builder%2522%252C%2522builder%2522%253A%257B%2522queryData%2522%253A%255B%257B%2522queryName%2522%253A%2522A%2522%252C%2522stepInterval%2522%253A60%252C%2522dataSource%2522%253A%2522traces%2522%252C%2522aggregateOperator%2522%253A%2522noop%2522%252C%2522aggregateAttribute%2522%253A%257B%2522key%2522%253A%2522%2522%252C%2522dataType%2522%253A%2522%2522%252C%2522type%2522%253A%2522%2522%252C%2522isColumn%2522%253Afalse%252C%2522isJSON%2522%253Afalse%257D%252C%2522expression%2522%253A%2522A%2522%252C%2522disabled%2522%253Afalse%252C%2522limit%2522%253A0%252C%2522offset%2522%253A0%252C%2522pageSize%2522%253A0%252C%2522ShiftBy%2522%253A0%252C%2522IsAnomaly%2522%253Afalse%252C%2522QueriesUsedInFormula%2522%253Anull%252C%2522filter%2522%253A%257B%2522expression%2522%253A%2522service.name%2BEXISTS%2522%257D%257D%255D%252C%2522queryFormulas%2522%253A%255B%255D%257D%257D&timeRange=%7B%22start%22%3A1753526700000000000%2C%22end%22%3A1753527000000000000%2C%22pageSize%22%3A100%7D&startTime=1753526700000000000&endTime=1753527000000000000&options=%7B%22maxLines%22%3A0%2C%22format%22%3A%22%22%2C%22selectColumns%22%3Anull%7D")
|
||||
}
|
||||
|
||||
func TestPrepareParamsForTraces(t *testing.T) {
|
||||
func TestPrepareLinksToTraces(t *testing.T) {
|
||||
postableRule := ruletypes.PostableRule{
|
||||
AlertName: "Links to traces test",
|
||||
AlertType: ruletypes.AlertTypeTraces,
|
||||
@@ -367,18 +360,15 @@ func TestPrepareParamsForTraces(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
externalURL := mustParseURL(t, "http://localhost:8000")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
|
||||
ts := time.UnixMilli(1705469040000)
|
||||
|
||||
params := rule.prepareParamsForTraces(context.Background(), ts, ruletypes.Labels{}).Encode()
|
||||
assert.Contains(t, params, "&timeRange=%7B%22start%22%3A1705468620000000000%2C%22end%22%3A1705468920000000000%2C%22pageSize%22%3A100%7D")
|
||||
assert.Contains(t, params, "&startTime=1705468620000000000")
|
||||
assert.Contains(t, params, "&endTime=1705468920000000000")
|
||||
link := rule.prepareLinksToTraces(context.Background(), ts, ruletypes.Labels{})
|
||||
assert.Contains(t, link, "&timeRange=%7B%22start%22%3A1705468620000000000%2C%22end%22%3A1705468920000000000%2C%22pageSize%22%3A100%7D&startTime=1705468620000000000&endTime=1705468920000000000")
|
||||
}
|
||||
|
||||
func TestThresholdRuleLabelNormalization(t *testing.T) {
|
||||
@@ -454,8 +444,7 @@ func TestThresholdRuleLabelNormalization(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8000")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
assert.NoError(t, err)
|
||||
|
||||
values := c.values
|
||||
@@ -652,8 +641,7 @@ func TestThresholdRuleUnitCombinations(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8000")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger, externalURL)
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
@@ -760,8 +748,7 @@ func TestThresholdRuleNoData(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger, externalURL)
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger)
|
||||
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
@@ -866,8 +853,7 @@ func TestThresholdRuleTracesLink(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger, externalURL)
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
@@ -984,8 +970,7 @@ func TestThresholdRuleLogsLink(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger, externalURL)
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger)
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
@@ -1164,8 +1149,7 @@ func TestMultipleThresholdRule(t *testing.T) {
|
||||
"summary": "The rule threshold is set to {{$threshold}}, and the observed metric value is {{$value}}",
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger, externalURL)
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, querier, logger)
|
||||
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
@@ -1311,8 +1295,7 @@ func TestThresholdRuleEval_SendUnmatchedBypassesRecovery(t *testing.T) {
|
||||
}
|
||||
|
||||
logger := instrumentationtest.New().Logger()
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
require.NoError(t, err)
|
||||
|
||||
now := time.Now()
|
||||
@@ -1554,8 +1537,7 @@ func runEvalTests(t *testing.T, postableRule ruletypes.PostableRule, testCases [
|
||||
Spec: thresholds,
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
return
|
||||
@@ -1662,8 +1644,7 @@ func runMultiThresholdEvalTests(t *testing.T, postableRule ruletypes.PostableRul
|
||||
Spec: thresholds,
|
||||
}
|
||||
|
||||
externalURL := mustParseURL(t, "http://localhost:8080")
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, externalURL, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, nil, logger, WithEvalDelay(valuer.MustParseTextDuration("2m")))
|
||||
if err != nil {
|
||||
assert.NoError(t, err)
|
||||
return
|
||||
@@ -1950,7 +1931,6 @@ func TestThresholdEval_RequireMinPoints(t *testing.T) {
|
||||
&postableRule,
|
||||
querier,
|
||||
logger,
|
||||
mustParseURL(t, "http://localhost:8080"),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
t.Run(fmt.Sprintf("%d, %s", idx, c.description), func(t *testing.T) {
|
||||
|
||||
@@ -2,7 +2,10 @@ package ruletypes
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
@@ -188,3 +191,35 @@ func (rc *RuleCondition) String() string {
|
||||
return string(data)
|
||||
}
|
||||
|
||||
// PrepareRuleGeneratorURL creates an appropriate url for the rule. The URL is
|
||||
// sent in Slack messages as well as to other systems and allows backtracking
|
||||
// to the rule definition from the third party systems.
|
||||
func PrepareRuleGeneratorURL(ruleID string, source string) string {
|
||||
if source == "" {
|
||||
return source
|
||||
}
|
||||
|
||||
// check if source is a valid url
|
||||
parsedSource, err := url.Parse(source)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
// since we capture window.location when a new rule is created
|
||||
// we end up with rulesource host:port/alerts/new. in this case
|
||||
// we want to replace new with rule id parameter
|
||||
|
||||
hasNew := strings.LastIndex(source, "new")
|
||||
if hasNew > -1 {
|
||||
ruleURL := fmt.Sprintf("%sedit?ruleId=%s", source[0:hasNew], ruleID)
|
||||
return ruleURL
|
||||
}
|
||||
|
||||
// The source contains the encoded query, start and end time
|
||||
// and other parameters. We don't want to include them in the generator URL
|
||||
// mainly to keep the URL short and lower the alert body contents
|
||||
// The generator URL with /alerts/edit?ruleId= is enough
|
||||
if parsedSource.Port() != "" {
|
||||
return fmt.Sprintf("%s://%s:%s/alerts/edit?ruleId=%s", parsedSource.Scheme, parsedSource.Hostname(), parsedSource.Port(), ruleID)
|
||||
}
|
||||
return fmt.Sprintf("%s://%s/alerts/edit?ruleId=%s", parsedSource.Scheme, parsedSource.Hostname(), ruleID)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user