Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[coordinator] Add config option to make rollup rules untimed #3343

Merged
merged 7 commits into from
Mar 10, 2021
Merged

[coordinator] Add config option to make rollup rules untimed #3343

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e139fc8
[coordinator] add config option to make rollup rules untimed
shreyassrivatsan Mar 9, 2021
a55fdf3
undo ci change
shreyassrivatsan Mar 9, 2021
3634cce
lint
shreyassrivatsan Mar 9, 2021
7f96428
change test
shreyassrivatsan Mar 9, 2021
8f50945
also drop timnestamp for the non-rolled up metric
shreyassrivatsan Mar 9, 2021
e79a476
use IsMappingRules
shreyassrivatsan Mar 10, 2021
14a4c4b
address comments
shreyassrivatsan Mar 10, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/cmd/services/m3coordinator/downsample/downsampler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,7 @@ func defaultMetricsAppenderOptions(opts DownsamplerOptions, agg agg) metricsAppe
metricTagsIteratorPool: agg.pools.metricTagsIteratorPool,
debugLogging: debugLogging,
logger: logger,
untimedRollups: agg.untimedRollups,
}
}

Expand Down
337 changes: 334 additions & 3 deletions src/cmd/services/m3coordinator/downsample/downsampler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1813,7 +1813,321 @@ func TestDownsamplerAggregationWithRulesConfigRollupRuleAndDropPolicyAndDropTime
"status_code": "500",
"endpoint": "/foo/bar",
},
values: []expectedValue{{value: 55}, {value: 39}},
values: []expectedValue{{value: 94}},
attributes: &storagemetadata.Attributes{
MetricsType: storagemetadata.AggregatedMetricsType,
Resolution: res,
Retention: ret,
},
},
},
},
})

// Test expected output
testDownsamplerAggregation(t, testDownsampler)
}

func TestDownsamplerAggregationWithRulesConfigRollupRuleUntimedRollups(t *testing.T) {
t.Parallel()

gaugeMetrics := []testGaugeMetric{
{
tags: map[string]string{
nameTag: "http_requests",
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
"not_rolled_up": "not_rolled_up_value_1",
},
timedSamples: []testGaugeMetricTimedSample{
{value: 42},
{value: 12, offset: 2 * time.Second},
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

interesting, so these will actually get rolled up into one data point even though the resolution is one second?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, so because timestamp gets stamped on the agg side, what you pass in is ignored.

I added another variety of test where we wait for the time before sending in the metric. That actually aggregates things into two datapoints.

},
expectDropTimestamp: true,
},
{
tags: map[string]string{
nameTag: "http_requests",
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
"not_rolled_up": "not_rolled_up_value_2",
},
timedSamples: []testGaugeMetricTimedSample{
{value: 13},
{value: 27, offset: 2 * time.Second},
},
expectDropTimestamp: true,
},
}
res := 1 * time.Second
ret := 30 * 24 * time.Hour
filter := fmt.Sprintf("%s:http_requests app:* status_code:* endpoint:*", nameTag)
testDownsampler := newTestDownsampler(t, testDownsamplerOptions{
untimedRollups: true,
rulesConfig: &RulesConfiguration{
MappingRules: []MappingRuleConfiguration{
{
Filter: "app:nginx*",
Aggregations: []aggregation.Type{aggregation.Max},
StoragePolicies: []StoragePolicyConfiguration{
{
Resolution: 1 * time.Second,
Retention: 30 * 24 * time.Hour,
},
},
},
},
RollupRules: []RollupRuleConfiguration{
{
Filter: filter,
Transforms: []TransformConfiguration{
{
Rollup: &RollupOperationConfiguration{
MetricName: "http_requests_by_status_code",
GroupBy: []string{"app", "status_code", "endpoint"},
Aggregations: []aggregation.Type{aggregation.Sum},
},
},
},
StoragePolicies: []StoragePolicyConfiguration{
{
Resolution: res,
Retention: ret,
},
},
},
},
},
ingest: &testDownsamplerOptionsIngest{
gaugeMetrics: gaugeMetrics,
},
expect: &testDownsamplerOptionsExpect{
writes: []testExpectedWrite{
{
tags: map[string]string{
nameTag: "http_requests_by_status_code",
string(rollupTagName): string(rollupTagValue),
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
},
values: []expectedValue{{value: 94}},
attributes: &storagemetadata.Attributes{
MetricsType: storagemetadata.AggregatedMetricsType,
Resolution: res,
Retention: ret,
},
},
},
},
})

// Test expected output
testDownsamplerAggregation(t, testDownsampler)
}

func TestDownsamplerAggregationWithRulesConfigRollupRuleUntimedRollupsWaitForOffset(t *testing.T) {
t.Parallel()

gaugeMetrics := []testGaugeMetric{
{
tags: map[string]string{
nameTag: "http_requests",
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
"not_rolled_up": "not_rolled_up_value_1",
},
timedSamples: []testGaugeMetricTimedSample{
{value: 42},
},
expectDropPolicyApplied: true,
expectDropTimestamp: true,
},
{
tags: map[string]string{
nameTag: "http_requests",
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
"not_rolled_up": "not_rolled_up_value_2",
},
timedSamples: []testGaugeMetricTimedSample{
{value: 12, offset: 2 * time.Second},
},
expectDropPolicyApplied: true,
expectDropTimestamp: true,
},
{
tags: map[string]string{
nameTag: "http_requests",
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
"not_rolled_up": "not_rolled_up_value_3",
},
timedSamples: []testGaugeMetricTimedSample{
{value: 13},
},
expectDropPolicyApplied: true,
expectDropTimestamp: true,
},
}
res := 1 * time.Second
ret := 30 * 24 * time.Hour
filter := fmt.Sprintf("%s:http_requests app:* status_code:* endpoint:*", nameTag)
testDownsampler := newTestDownsampler(t, testDownsamplerOptions{
waitForOffset: true,
untimedRollups: true,
rulesConfig: &RulesConfiguration{
MappingRules: []MappingRuleConfiguration{
{
Filter: filter,
Drop: true,
},
},
RollupRules: []RollupRuleConfiguration{
{
Filter: filter,
Transforms: []TransformConfiguration{
{
Rollup: &RollupOperationConfiguration{
MetricName: "http_requests_by_status_code",
GroupBy: []string{"app", "status_code", "endpoint"},
Aggregations: []aggregation.Type{aggregation.Sum},
},
},
},
StoragePolicies: []StoragePolicyConfiguration{
{
Resolution: res,
Retention: ret,
},
},
},
},
},
ingest: &testDownsamplerOptionsIngest{
gaugeMetrics: gaugeMetrics,
},
expect: &testDownsamplerOptionsExpect{
writes: []testExpectedWrite{
{
tags: map[string]string{
nameTag: "http_requests_by_status_code",
string(rollupTagName): string(rollupTagValue),
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
},
values: []expectedValue{{value: 42}, {value: 25}},
attributes: &storagemetadata.Attributes{
MetricsType: storagemetadata.AggregatedMetricsType,
Resolution: res,
Retention: ret,
},
},
},
},
})

// Test expected output
testDownsamplerAggregation(t, testDownsampler)
}

func TestDownsamplerAggregationWithRulesConfigRollupRuleRollupLaterUntimedRollups(t *testing.T) {
t.Parallel()

gaugeMetrics := []testGaugeMetric{
{
tags: map[string]string{
nameTag: "http_requests",
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
"not_rolled_up": "not_rolled_up_value_1",
},
timedSamples: []testGaugeMetricTimedSample{
{value: 42},
{value: 12, offset: 2 * time.Second},
},
expectDropTimestamp: true,
},
{
tags: map[string]string{
nameTag: "http_requests",
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
"not_rolled_up": "not_rolled_up_value_2",
},
timedSamples: []testGaugeMetricTimedSample{
{value: 13},
{value: 27, offset: 2 * time.Second},
},
expectDropTimestamp: true,
},
}
res := 1 * time.Second
ret := 30 * 24 * time.Hour
filter := fmt.Sprintf("%s:http_requests app:* status_code:* endpoint:*", nameTag)
testDownsampler := newTestDownsampler(t, testDownsamplerOptions{
untimedRollups: true,
rulesConfig: &RulesConfiguration{
MappingRules: []MappingRuleConfiguration{
{
Filter: "app:nginx*",
Aggregations: []aggregation.Type{aggregation.Max},
StoragePolicies: []StoragePolicyConfiguration{
{
Resolution: 1 * time.Second,
Retention: 30 * 24 * time.Hour,
},
},
},
},
RollupRules: []RollupRuleConfiguration{
{
Filter: filter,
Transforms: []TransformConfiguration{
{
Transform: &TransformOperationConfiguration{
Type: transformation.Add,
},
},
{
Rollup: &RollupOperationConfiguration{
MetricName: "http_requests_by_status_code",
GroupBy: []string{"app", "status_code", "endpoint"},
Aggregations: []aggregation.Type{aggregation.Sum},
},
},
},
StoragePolicies: []StoragePolicyConfiguration{
{
Resolution: res,
Retention: ret,
},
},
},
},
},
ingest: &testDownsamplerOptionsIngest{
gaugeMetrics: gaugeMetrics,
},
expect: &testDownsamplerOptionsExpect{
writes: []testExpectedWrite{
{
tags: map[string]string{
nameTag: "http_requests_by_status_code",
string(rollupTagName): string(rollupTagValue),
"app": "nginx_edge",
"status_code": "500",
"endpoint": "/foo/bar",
},
values: []expectedValue{{value: 39}},
attributes: &storagemetadata.Attributes{
MetricsType: storagemetadata.AggregatedMetricsType,
Resolution: res,
Expand Down Expand Up @@ -2644,7 +2958,14 @@ func testDownsamplerAggregationIngest(
if sample.offset > 0 {
sample.time = sample.time.Add(sample.offset)
}
err = samplesAppender.AppendCounterSample(sample.time, sample.value, nil)
if testOpts.waitForOffset {
time.Sleep(sample.offset)
}
if samplesAppenderResult.ShouldDropTimestamp {
err = samplesAppender.AppendUntimedCounterSample(sample.value, nil)
} else {
err = samplesAppender.AppendCounterSample(sample.time, sample.value, nil)
}
require.NoError(t, err)
}
}
Expand Down Expand Up @@ -2674,7 +2995,14 @@ func testDownsamplerAggregationIngest(
if sample.offset > 0 {
sample.time = sample.time.Add(sample.offset)
}
err = samplesAppender.AppendGaugeSample(sample.time, sample.value, nil)
if testOpts.waitForOffset {
time.Sleep(sample.offset)
}
if samplesAppenderResult.ShouldDropTimestamp {
err = samplesAppender.AppendUntimedGaugeSample(sample.value, nil)
} else {
err = samplesAppender.AppendGaugeSample(sample.time, sample.value, nil)
}
require.NoError(t, err)
}
}
Expand Down Expand Up @@ -2718,6 +3046,8 @@ type testDownsamplerOptions struct {
clockOpts clock.Options
instrumentOpts instrument.Options
identTag string
untimedRollups bool
waitForOffset bool

// Options for the test
autoMappingRules []m3.ClusterNamespaceOptions
Expand Down Expand Up @@ -2808,6 +3138,7 @@ func newTestDownsampler(t *testing.T, opts testDownsamplerOptions) testDownsampl
cfg.Rules = opts.rulesConfig
}
cfg.Matcher = opts.matcherConfig
cfg.UntimedRollups = opts.untimedRollups

instance, err := cfg.NewDownsampler(DownsamplerOptions{
Storage: storage,
Expand Down
Loading