hanzo/logs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Default ingestion policy mappings merged with per-tenant mappings (#18926)

Browse Source
This commit is contained in:
Salva Corts
2025-08-21 13:12:57 +02:00
committed by GitHub
parent 43dfc9740e
commit 3b2498b896
7 changed files with 247 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/loki/main.go
+5
View File
@@ -44,6 +44,11 @@ func main() {
// Set the global OTLP config which is needed in per tenant otlp config
config.LimitsConfig.SetGlobalOTLPConfig(config.Distributor.OTLPConfig)
// Set the default policy stream mappings which are needed in per tenant policy stream mappings
if err := config.LimitsConfig.SetDefaultPolicyStreamMapping(config.Distributor.DefaultPolicyStreamMappings); err != nil {
level.Error(util_log.Logger).Log("msg", "failed to set default policy stream mappings", "err", err.Error())
exit(1)
}
// This global is set to the config passed into the last call to `NewOverrides`. If we don't
// call it atleast once, the defaults are set to an empty struct.
// We call it with the flag values so that the config file unmarshalling only overrides the values set in the config.
docs/sources/shared/configuration.md
+3
View File
@@ -2911,6 +2911,9 @@ otlp_config:
# CLI flag: -distributor.otlp.default_resource_attributes_as_index_labels
[default_resource_attributes_as_index_labels: <list of strings> | default = [service.name service.namespace service.instance.id deployment.environment deployment.environment.name cloud.region cloud.availability_zone k8s.cluster.name k8s.namespace.name k8s.pod.name k8s.container.name container.name k8s.replicaset.name k8s.deployment.name k8s.statefulset.name k8s.daemonset.name k8s.cronjob.name k8s.job.name]]
# Default policy stream mappings that are merged with per-tenant mappings.
[default_policy_stream_mappings: <map of string to list of PriorityStreams>]
# Enable writes to Kafka during Push requests.
# CLI flag: -distributor.kafka-writes-enabled
[kafka_writes_enabled: <boolean> | default = false]
integration/cluster/cluster.go
+3
View File
@@ -441,6 +441,9 @@ func (c *Component) run() error {
}
config.LimitsConfig.SetGlobalOTLPConfig(config.Distributor.OTLPConfig)
if err := config.LimitsConfig.SetDefaultPolicyStreamMapping(config.Distributor.DefaultPolicyStreamMappings); err != nil {
return err
}
var err error
c.loki, err = loki.New(config.Config)
if err != nil {
pkg/distributor/distributor.go
+3
View File
@@ -100,6 +100,9 @@ type Config struct {
OTLPConfig push.GlobalOTLPConfig `yaml:"otlp_config"`
// DefaultPolicyStreamMappings contains the default policy stream mappings that are merged with per-tenant mappings.
DefaultPolicyStreamMappings validation.PolicyStreamMapping `yaml:"default_policy_stream_mappings" doc:"description=Default policy stream mappings that are merged with per-tenant mappings."`
KafkaEnabled bool `yaml:"kafka_writes_enabled"`
IngesterEnabled bool `yaml:"ingester_writes_enabled"`
IngestLimitsEnabled bool `yaml:"ingest_limits_enabled"`
pkg/validation/ingestion_policies.go
+51
View File
@@ -92,3 +92,54 @@ func (p *PolicyStreamMapping) PolicyFor(lbs labels.Labels) []string {
return policies
}
// ApplyDefaultPolicyStreamMappings applies default policy stream mappings to the current mapping.
// The defaults are merged with the existing mappings, with existing mappings taking precedence.
func (p *PolicyStreamMapping) ApplyDefaultPolicyStreamMappings(defaults PolicyStreamMapping) error {
if defaults == nil {
return nil
}
// If the current mapping is nil, initialize it
if *p == nil {
*p = make(PolicyStreamMapping)
}
// Merge defaults with existing mappings
for policyName, defaultStreams := range defaults {
if existingStreams, exists := (*p)[policyName]; exists {
// If the policy already exists, merge the streams
// We need to check for duplicates based on selector to avoid adding the same stream twice
existingSelectors := make(map[string]bool)
for _, stream := range existingStreams {
existingSelectors[stream.Selector] = true
}
// Add default streams that don't already exist
for _, defaultStream := range defaultStreams {
if !existingSelectors[defaultStream.Selector] {
existingStreams = append(existingStreams, defaultStream)
}
}
(*p)[policyName] = existingStreams
} else {
// If the policy doesn't exist, copy all default streams
streamsCopy := make([]*PriorityStream, len(defaultStreams))
for i, stream := range defaultStreams {
streamsCopy[i] = &PriorityStream{
Priority: stream.Priority,
Selector: stream.Selector,
Matchers: stream.Matchers,
}
}
(*p)[policyName] = streamsCopy
}
}
// Re-validate after merging to ensure proper sorting. The defaults are already validated
// so this should not fail, but playing it safe here and returning the error.
if err := p.Validate(); err != nil {
return fmt.Errorf("validation failed after merging with the defaults: %w", err)
}
return nil
}
pkg/validation/ingestion_policies_test.go
+168
View File
@@ -116,3 +116,171 @@ func Test_PolicyStreamMapping_PolicyFor(t *testing.T) {
// Matches policy7 and policy8 which have the same priority.
require.Equal(t, []string{"policy7", "policy8"}, mapping.PolicyFor(labels.FromStrings("env", "prod")))
}
func TestPolicyStreamMapping_ApplyDefaultPolicyStreamMappings(t *testing.T) {
tests := []struct {
name string
existing PolicyStreamMapping
defaults PolicyStreamMapping
expected PolicyStreamMapping
expectedError bool
}{
{
name: "nil existing, nil defaults",
existing: nil,
defaults: nil,
expected: nil,
},
{
name: "nil existing, with defaults",
existing: nil,
defaults: PolicyStreamMapping{
"policy1": {
{Priority: 1, Selector: "{app=\"test\"}"},
},
},
expected: PolicyStreamMapping{
"policy1": {
{Priority: 1, Selector: "{app=\"test\"}"},
},
},
},
{
name: "existing with defaults, no overlap",
existing: PolicyStreamMapping{
"policy1": {
{Priority: 2, Selector: "{app=\"existing\"}"},
},
},
defaults: PolicyStreamMapping{
"policy2": {
{Priority: 1, Selector: "{app=\"default\"}"},
},
},
expected: PolicyStreamMapping{
"policy1": {
{Priority: 2, Selector: "{app=\"existing\"}"},
},
"policy2": {
{Priority: 1, Selector: "{app=\"default\"}"},
},
},
},
{
name: "existing with defaults, with overlap - existing takes precedence",
existing: PolicyStreamMapping{
"policy1": {
{Priority: 2, Selector: "{app=\"existing\"}"},
},
},
defaults: PolicyStreamMapping{
"policy1": {
{Priority: 1, Selector: "{app=\"existing\"}"}, // Same selector, different priority
},
"policy2": {
{Priority: 1, Selector: "{app=\"default\"}"},
},
},
expected: PolicyStreamMapping{
"policy1": {
{Priority: 2, Selector: "{app=\"existing\"}"}, // Existing priority preserved
},
"policy2": {
{Priority: 1, Selector: "{app=\"default\"}"},
},
},
},
{
name: "existing with defaults, merge different selectors",
existing: PolicyStreamMapping{
"policy1": {
{Priority: 2, Selector: "{app=\"existing\"}"},
},
},
defaults: PolicyStreamMapping{
"policy1": {
{Priority: 1, Selector: "{app=\"default\"}"}, // Different selector
},
},
expected: PolicyStreamMapping{
"policy1": {
{Priority: 2, Selector: "{app=\"existing\"}"},
{Priority: 1, Selector: "{app=\"default\"}"},
},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Create a copy of existing for the test
var existingCopy PolicyStreamMapping
if tt.existing != nil {
existingCopy = make(PolicyStreamMapping)
for k, v := range tt.existing {
streamsCopy := make([]*PriorityStream, len(v))
for i, stream := range v {
streamsCopy[i] = &PriorityStream{
Priority: stream.Priority,
Selector: stream.Selector,
Matchers: stream.Matchers,
}
}
existingCopy[k] = streamsCopy
}
}
// Apply defaults
err := existingCopy.ApplyDefaultPolicyStreamMappings(tt.defaults)
if tt.expectedError {
require.Error(t, err)
return
}
require.NoError(t, err)
// Validate the result
if tt.expected == nil {
require.Nil(t, existingCopy)
} else {
require.NotNil(t, existingCopy)
require.Equal(t, len(tt.expected), len(existingCopy))
for policyName, expectedStreams := range tt.expected {
actualStreams, exists := existingCopy[policyName]
require.True(t, exists, "Policy %s should exist", policyName)
require.Equal(t, len(expectedStreams), len(actualStreams))
// Check each stream
for i, expectedStream := range expectedStreams {
require.Less(t, i, len(actualStreams))
actualStream := actualStreams[i]
require.Equal(t, expectedStream.Priority, actualStream.Priority)
require.Equal(t, expectedStream.Selector, actualStream.Selector)
}
}
}
})
}
}
func TestPolicyStreamMapping_ApplyDefaultPolicyStreamMappings_Validation(t *testing.T) {
// Test that validation is called after merging
existing := PolicyStreamMapping{
"policy1": {
{Priority: 2, Selector: "{app=\"existing\"}"},
},
}
defaults := PolicyStreamMapping{
"policy1": {
{Priority: 1, Selector: "{app=\"default\"}"},
},
}
// This should not panic and should call Validate()
err := existing.ApplyDefaultPolicyStreamMappings(defaults)
require.NoError(t, err)
// Verify the result is valid
require.NoError(t, existing.Validate())
}
pkg/validation/limits.go
+14
View File
@@ -240,6 +240,10 @@ type Limits struct {
PolicyEnforcedLabels map[string][]string `yaml:"policy_enforced_labels" json:"policy_enforced_labels" category:"experimental" doc:"description=Map of policies to enforced labels. The policy '*' is the global policy, which is applied to all streams and can be extended by other policies. Example:\n policy_enforced_labels: \n policy1: \n - label1 \n - label2 \n policy2: \n - label3 \n - label4\n '*':\n - label5"`
PolicyStreamMapping PolicyStreamMapping `yaml:"policy_stream_mapping" json:"policy_stream_mapping" category:"experimental" doc:"description=Map of policies to stream selectors with a priority. Experimental. Example:\n policy_stream_mapping: \n finance: \n - selector: '{namespace=\"prod\", container=\"billing\"}' \n priority: 2 \n ops: \n - selector: '{namespace=\"prod\", container=\"ops\"}' \n priority: 1 \n staging: \n - selector: '{namespace=\"staging\"}' \n priority: 1"`
// DefaultPolicyStreamMapping contains the default policy stream mappings that are merged with per-tenant mappings.
// This field is not exposed in YAML/JSON as it's set programmatically.
DefaultPolicyStreamMapping PolicyStreamMapping `yaml:"-" json:"-"`
IngestionPartitionsTenantShardSize int `yaml:"ingestion_partitions_tenant_shard_size" json:"ingestion_partitions_tenant_shard_size" category:"experimental"`
ShardAggregations []string `yaml:"shard_aggregations,omitempty" json:"shard_aggregations,omitempty" doc:"description=List of LogQL vector and range aggregations that should be sharded."`
@@ -504,6 +508,12 @@ func (l *Limits) SetGlobalOTLPConfig(cfg push.GlobalOTLPConfig) {
l.OTLPConfig.ApplyGlobalOTLPConfig(cfg)
}
// SetDefaultPolicyStreamMapping sets DefaultPolicyStreamMapping which is used while unmarshaling per-tenant policy stream mappings to use the default mappings.
func (l *Limits) SetDefaultPolicyStreamMapping(cfg PolicyStreamMapping) error {
l.DefaultPolicyStreamMapping = cfg
return l.PolicyStreamMapping.ApplyDefaultPolicyStreamMappings(cfg)
}
// UnmarshalYAML implements the yaml.Unmarshaler interface.
func (l *Limits) UnmarshalYAML(unmarshal func(interface{}) error) error {
// We want to set c to the defaults and then overwrite it with the input.
@@ -528,6 +538,10 @@ func (l *Limits) UnmarshalYAML(unmarshal func(interface{}) error) error {
if defaultLimits != nil {
// apply relevant bits from global otlp config
l.OTLPConfig.ApplyGlobalOTLPConfig(defaultLimits.GlobalOTLPConfig)
// apply default policy stream mappings
if err := l.PolicyStreamMapping.ApplyDefaultPolicyStreamMappings(defaultLimits.DefaultPolicyStreamMapping); err != nil {
return errors.Wrap(err, "applying default policy stream mappings")
}
}
return nil
}