Skip to content

vmdistributed: support trafficMode #2100

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

Open
AndrewChubatiuk wants to merge 1 commit into
base: master
Choose a base branch
from
Open

vmdistributed: support trafficMode #2100

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
12 changes: 12 additions & 0 deletions api/operator/v1alpha1/vmdistributed_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,11 +84,23 @@ type VMDistributedZoneCommon struct {
UpdatePause *metav1.Duration `json:"updatePause,omitempty"`
}

type VMDistributedTrafficMode string

const (
VMDistributedTrafficModeReadWrite VMDistributedTrafficMode = "read-write"
VMDistributedTrafficModeReadOnly VMDistributedTrafficMode = "read-only"
VMDistributedTrafficModeWriteOnly VMDistributedTrafficMode = "write-only"
VMDistributedTrafficModeMaintenance VMDistributedTrafficMode = "maintenance"
)

// +k8s:openapi-gen=true
// VMDistributedZone defines items within a single zone to update.
type VMDistributedZone struct {
// Name defines a name of zone, which can be used in zoneCommon spec as %ZONE%
Name string `json:"name"`
// TrafficMode defines allowed traffic mode for a zone: read-only, write-only, read-write, maintenance
// +kubebuilder:validation:Enum=read-only;write-only;read-write;maintenance
TrafficMode VMDistributedTrafficMode `json:"trafficMode,omitempty"`
// VMCluster defines a new inline or referencing existing one VMCluster
// +optional
VMCluster VMDistributedZoneCluster `json:"vmcluster,omitempty"`
Expand Down
7 changes: 7 additions & 0 deletions config/crd/overlay/crd.descriptionless.yaml
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

9 changes: 9 additions & 0 deletions config/crd/overlay/crd.yaml
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions docs/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ aliases:
## tip

* FEATURE: [vmauth](https://docs.victoriametrics.com/operator/resources/vmauth): previously VMAuth could read configuration only from predefined locations; now VMAuth supports arbitrary filesystem access configuration, allowing users to reference required files directly and reducing configuration workarounds. See [#899](https://github.com/VictoriaMetrics/operator/issues/899).
* FEATURE: [vmdistributed](https://docs.victoriametrics.com/operator/resources/vmdistributed): introduce `spec.zones[*].trafficMode` property, which allows disable read, write or whole traffic to a zone. See [#1995](https://github.com/VictoriaMetrics/operator/issues/1995).

* BUGFIX: [converter](https://docs.victoriametrics.com/operator/integrations/prometheus/#objects-conversion): disable all prometheus controllers if CRD group was not found. See [#2838](https://github.com/VictoriaMetrics/helm-charts/issues/2838).

Expand Down
7 changes: 7 additions & 0 deletions docs/api.md
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

7 changes: 6 additions & 1 deletion internal/controller/operator/factory/build/defaults.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@ func AddDefaults(scheme *runtime.Scheme) {

func addVMDistributedDefaults(objI any) {
cr := objI.(*vmv1alpha1.VMDistributed)

if cr.Spec.ZoneCommon.ReadyTimeout == nil {
cr.Spec.ZoneCommon.ReadyTimeout = &metav1.Duration{
Duration: 5 * time.Minute,
Expand All @@ -63,6 +62,12 @@ func addVMDistributedDefaults(objI any) {
Duration: 1 * time.Minute,
}
}
for i := range cr.Spec.Zones {
z := &cr.Spec.Zones[i]
if len(z.TrafficMode) == 0 {
z.TrafficMode = vmv1alpha1.VMDistributedTrafficModeReadWrite
}
}
if cr.Spec.License.IsProvided() {
if !cr.Spec.VMAuth.Spec.License.IsProvided() {
cr.Spec.VMAuth.Spec.License = cr.Spec.License.DeepCopy()
Expand Down
19 changes: 16 additions & 3 deletions internal/controller/operator/factory/vmdistributed/vmauth.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ func vmAgentTargetRef(vmAgents []*vmv1beta1.VMAgent, owner *metav1.OwnerReferenc
return ref
}

func buildVMAuthLB(cr *vmv1alpha1.VMDistributed, vmAgents []*vmv1beta1.VMAgent, vmClusters []*vmv1beta1.VMCluster, excludeIds ...int) *vmv1beta1.VMAuth {
func buildVMAuthLB(cr *vmv1alpha1.VMDistributed, vmAgents []*vmv1beta1.VMAgent, vmClusters []*vmv1beta1.VMCluster, trafficModes []vmv1alpha1.VMDistributedTrafficMode, excludeIds ...int) *vmv1beta1.VMAuth {
if !ptr.Deref(cr.Spec.VMAuth.Enabled, true) || build.IsControllerDisabled("VMAuth") {
return nil
}
Expand All @@ -108,10 +108,23 @@ func buildVMAuthLB(cr *vmv1alpha1.VMDistributed, vmAgents []*vmv1beta1.VMAgent,
},
Spec: *cr.Spec.VMAuth.Spec.DeepCopy(),
}
writeExcludeIds := slices.Clone(excludeIds)
readExcludeIds := slices.Clone(excludeIds)
for i, mode := range trafficModes {
switch mode {
case vmv1alpha1.VMDistributedTrafficModeReadOnly:
writeExcludeIds = append(writeExcludeIds, i)
case vmv1alpha1.VMDistributedTrafficModeWriteOnly:
readExcludeIds = append(readExcludeIds, i)
case vmv1alpha1.VMDistributedTrafficModeMaintenance:
writeExcludeIds = append(writeExcludeIds, i)
readExcludeIds = append(readExcludeIds, i)
}
}
var targetRefs []vmv1beta1.TargetRef
owner := cr.AsOwner()
targetRefs = append(targetRefs, vmAgentTargetRef(vmAgents, &owner, excludeIds...))
targetRefs = append(targetRefs, vmClusterTargetRef(vmClusters, &owner, excludeIds...))
targetRefs = append(targetRefs, vmAgentTargetRef(vmAgents, &owner, writeExcludeIds...))
targetRefs = append(targetRefs, vmClusterTargetRef(vmClusters, &owner, readExcludeIds...))
vmAuth.Spec.DefaultTargetRefs = targetRefs
return &vmAuth
}
2 changes: 1 addition & 1 deletion internal/controller/operator/factory/vmdistributed/vmauth_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,7 @@ func TestBuildVMAuthLBZoneOrder(t *testing.T) {

f := func(o opts) {
t.Helper()
vmAuth := buildVMAuthLB(cr, o.agents, o.clusters, o.excludeIds...)
vmAuth := buildVMAuthLB(cr, o.agents, o.clusters, nil, o.excludeIds...)
assert.NotNil(t, vmAuth)
assert.Len(t, vmAuth.Spec.DefaultTargetRefs, 2)

Expand Down
22 changes: 11 additions & 11 deletions internal/controller/operator/factory/vmdistributed/vmdistributed_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -276,7 +276,7 @@ func TestCreateOrUpdate(t *testing.T) {
}
},
validate: func(ctx context.Context, rclient client.Client, d *testData) {
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters)
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters, d.zones.trafficModes)
owner := d.cr.AsOwner()
assert.NoError(t, reconcile.VMAuth(ctx, rclient, vmAuth, nil, &owner))
},
Expand All @@ -291,7 +291,7 @@ func TestCreateOrUpdate(t *testing.T) {
},
preRun: func(c client.Client, d *testData) {
clusters := []*vmv1beta1.VMCluster{d.zones.vmclusters[0]}
lb := buildVMAuthLB(d.cr, d.zones.vmagents, clusters)
lb := buildVMAuthLB(d.cr, d.zones.vmagents, clusters, nil)
c.Scheme().Default(lb)
assert.NoError(t, c.Create(context.TODO(), lb))
},
Expand All @@ -316,7 +316,7 @@ func TestCreateOrUpdate(t *testing.T) {
LogLevel: "INFO",
}
clusters := []*vmv1beta1.VMCluster{d.zones.vmclusters[0]}
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, clusters)
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, clusters, nil)
owner := d.cr.AsOwner()
assert.NoError(t, reconcile.VMAuth(ctx, rclient, vmAuth, nil, &owner))
},
Expand All @@ -333,7 +333,7 @@ func TestCreateOrUpdate(t *testing.T) {
},
preRun: func(c client.Client, d *testData) {
clusters := []*vmv1beta1.VMCluster{d.zones.vmclusters[0]}
lb := buildVMAuthLB(d.cr, d.zones.vmagents, clusters)
lb := buildVMAuthLB(d.cr, d.zones.vmagents, clusters, nil)
c.Scheme().Default(lb)
assert.NoError(t, c.Create(context.TODO(), lb))
},
Expand All @@ -351,7 +351,7 @@ func TestCreateOrUpdate(t *testing.T) {
},
validate: func(ctx context.Context, rclient client.Client, d *testData) {
clusters := []*vmv1beta1.VMCluster{d.zones.vmclusters[0]}
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, clusters)
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, clusters, nil)
owner := d.cr.AsOwner()
assert.NoError(t, reconcile.VMAuth(ctx, rclient, vmAuth, nil, &owner))
},
Expand Down Expand Up @@ -389,7 +389,7 @@ func TestCreateOrUpdate(t *testing.T) {
}
},
validate: func(ctx context.Context, rclient client.Client, d *testData) {
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters)
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters, d.zones.trafficModes)
owner := d.cr.AsOwner()
assert.NoError(t, reconcile.VMAuth(ctx, rclient, vmAuth, nil, &owner))
var vmClusterObjs, vmAgentObjs []vmv1beta1.NamespacedName
Expand Down Expand Up @@ -451,7 +451,7 @@ func TestCreateOrUpdate(t *testing.T) {
},
preRun: func(c client.Client, d *testData) {
clusters := []*vmv1beta1.VMCluster{d.zones.vmclusters[0]}
lb := buildVMAuthLB(d.cr, d.zones.vmagents, clusters)
lb := buildVMAuthLB(d.cr, d.zones.vmagents, clusters, nil)
c.Scheme().Default(lb)
lb.OwnerReferences = nil
assert.NoError(t, c.Create(context.TODO(), lb))
Expand All @@ -478,7 +478,7 @@ func TestCreateOrUpdate(t *testing.T) {
},
validate: func(ctx context.Context, rclient client.Client, d *testData) {
clusters := []*vmv1beta1.VMCluster{d.zones.vmclusters[0]}
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, clusters)
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, clusters, nil)
owner := d.cr.AsOwner()
assert.NoError(t, reconcile.VMAuth(ctx, rclient, vmAuth, nil, &owner))
var got vmv1beta1.VMAuth
Expand Down Expand Up @@ -524,7 +524,7 @@ func TestCreateOrUpdate(t *testing.T) {
}
},
validate: func(ctx context.Context, rclient client.Client, d *testData) {
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters)
vmAuth := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters, d.zones.trafficModes)
owner := d.cr.AsOwner()
assert.NoError(t, reconcile.VMAuth(ctx, rclient, vmAuth, nil, &owner))
var vmClusterObjs, vmAgentObjs []vmv1beta1.NamespacedName
Expand Down Expand Up @@ -582,7 +582,7 @@ func TestCreateOrUpdate(t *testing.T) {
},
preRun: func(c client.Client, d *testData) {
// Create initial VMAuth backed by CRD targets
lb := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters)
lb := buildVMAuthLB(d.cr, d.zones.vmagents, d.zones.vmclusters, d.zones.trafficModes)
c.Scheme().Default(lb)
assert.NoError(t, c.Create(context.TODO(), lb))
},
Expand All @@ -608,7 +608,7 @@ func TestCreateOrUpdate(t *testing.T) {
},
validate: func(ctx context.Context, rclient client.Client, d *testData) {
// Simulate removing both backends by passing empty slices
vmAuth := buildVMAuthLB(d.cr, nil, nil)
vmAuth := buildVMAuthLB(d.cr, nil, nil, nil)
owner := d.cr.AsOwner()
assert.NoError(t, reconcile.VMAuth(ctx, rclient, vmAuth, nil, &owner))

Expand Down
23 changes: 15 additions & 8 deletions internal/controller/operator/factory/vmdistributed/zone.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,11 @@ import (
)

type zones struct {
httpClient *http.Client
vmagents []*vmv1beta1.VMAgent
vmclusters []*vmv1beta1.VMCluster
hasChanges []bool
httpClient *http.Client
vmagents []*vmv1beta1.VMAgent
vmclusters []*vmv1beta1.VMCluster
hasChanges []bool
trafficModes []vmv1alpha1.VMDistributedTrafficMode
}

func (zs *zones) Len() int {
Expand Down Expand Up @@ -66,6 +67,7 @@ func (zs *zones) Swap(i, j int) {
zs.vmagents[i], zs.vmagents[j] = zs.vmagents[j], zs.vmagents[i]
zs.vmclusters[i], zs.vmclusters[j] = zs.vmclusters[j], zs.vmclusters[i]
zs.hasChanges[i], zs.hasChanges[j] = zs.hasChanges[j], zs.hasChanges[i]
zs.trafficModes[i], zs.trafficModes[j] = zs.trafficModes[j], zs.trafficModes[i]
}

// getZones builds desired zones
Expand All @@ -74,9 +76,10 @@ func getZones(ctx context.Context, rclient client.Client, cr *vmv1alpha1.VMDistr
httpClient: &http.Client{
Timeout: httpTimeout,
},
vmagents: make([]*vmv1beta1.VMAgent, len(cr.Spec.Zones)),
vmclusters: make([]*vmv1beta1.VMCluster, len(cr.Spec.Zones)),
hasChanges: make([]bool, len(cr.Spec.Zones)),
vmagents: make([]*vmv1beta1.VMAgent, len(cr.Spec.Zones)),
vmclusters: make([]*vmv1beta1.VMCluster, len(cr.Spec.Zones)),
hasChanges: make([]bool, len(cr.Spec.Zones)),
trafficModes: make([]vmv1alpha1.VMDistributedTrafficMode, len(cr.Spec.Zones)),
}
for i := range cr.Spec.Zones {
z := &cr.Spec.Zones[i]
Expand All @@ -102,8 +105,12 @@ func getZones(ctx context.Context, rclient client.Client, cr *vmv1alpha1.VMDistr
prevClusterSpec := vmCluster.Spec
vmCluster.Spec = *vmClusterSpec
rclient.Scheme().Default(&vmCluster)
if (z.TrafficMode == vmv1alpha1.VMDistributedTrafficModeReadOnly || z.TrafficMode == vmv1alpha1.VMDistributedTrafficModeMaintenance) && vmCluster.Spec.VMInsert != nil {
vmCluster.Spec.VMInsert.ReplicaCount = ptr.To(int32(0))
Copy link
Copy Markdown
Collaborator

@vrutkovs vrutkovs Apr 24, 2026

Choose a reason for hiding this comment

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

Wouldn't that make vmagent start accumulating the queue for RO clusters? Also, needs a comment why we do this

}
zs.hasChanges[i] = !equality.Semantic.DeepEqual(&vmCluster.Spec, &prevClusterSpec)
zs.vmclusters[i] = &vmCluster
zs.trafficModes[i] = z.TrafficMode
}

for i := range cr.Spec.Zones {
Expand Down Expand Up @@ -238,7 +245,7 @@ func (zs *zones) updateLB(ctx context.Context, rclient client.Client, cr *vmv1al
if !ptr.Deref(cr.Spec.VMAuth.Enabled, true) {
return nil
}
vmAuth := buildVMAuthLB(cr, zs.vmagents, zs.vmclusters, excludeIds...)
vmAuth := buildVMAuthLB(cr, zs.vmagents, zs.vmclusters, zs.trafficModes, excludeIds...)
if vmAuth == nil {
return nil
}
Expand Down
7 changes: 4 additions & 3 deletions internal/controller/operator/factory/vmdistributed/zone_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,9 +260,10 @@ func TestZonesSorting(t *testing.T) {
f := func(o opts) {
t.Helper()
zs := &zones{
vmclusters: o.clusters,
vmagents: make([]*vmv1beta1.VMAgent, len(o.clusters)),
hasChanges: make([]bool, len(o.clusters)),
vmclusters: o.clusters,
vmagents: make([]*vmv1beta1.VMAgent, len(o.clusters)),
hasChanges: make([]bool, len(o.clusters)),
trafficModes: make([]vmv1alpha1.VMDistributedTrafficMode, len(o.clusters)),
}
for i := range o.clusters {
zs.vmagents[i] = &vmv1beta1.VMAgent{
Expand Down
Loading