Skip to content

Commit b7c3e91

Browse files
committed
epp: add multicluster feature gate and endpoint attribute
Multi-cluster mode is declared at the EPP level, behind the multicluster feature gate, rather than per endpoint. When the gate is on, the datalayer Runtime marks every endpoint it creates with a multicluster attribute, so the mark covers all discovery sources including endpoints that are not pods. Plugins read the mark through IsMultiCluster, holding only the endpoint and never the global config. It is a feature gate rather than a config section because multi-cluster routing is still maturing and most EPPs never use it. Promotion looks like moving it out of the feature gate into a dedicated multiCluster section on EndpointPickerConfig, once it matures or needs structured configuration (peer discovery, cluster identity). The Runtime consumes a single WithMultiCluster option today, so that promotion is a config-surface change, not a behavior change. Signed-off-by: Sam Batschelet <sbatsche@redhat.com>
1 parent 44828e2 commit b7c3e91

4 files changed

Lines changed: 97 additions & 3 deletions

File tree

cmd/epp/runner/runner.go

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -685,6 +685,7 @@ func (r *Runner) parseConfigurationPhaseOne(ctx context.Context, opts *runserver
685685

686686
loader.RegisterFeatureGate(flowcontrol.FeatureGate, false)
687687
loader.RegisterFeatureGate(runserver.HAPopulateNonLeaderDatastoreFeatureGate, true)
688+
loader.RegisterFeatureGate(fwkdl.MultiClusterFeatureGate, false)
688689

689690
r.registerInTreePlugins()
690691

@@ -784,7 +785,8 @@ func (r *Runner) configureAndStartDatalayer(ctx context.Context, cfg *datalayer.
784785
}
785786

786787
func (r *Runner) setupMetricsCollection(opts *runserver.Options) datalayer.EndpointFactory {
787-
r.dlRuntime = datalayer.NewRuntime(opts.RefreshMetricsInterval)
788+
r.dlRuntime = datalayer.NewRuntime(opts.RefreshMetricsInterval,
789+
datalayer.WithMultiCluster(r.featureGates[fwkdl.MultiClusterFeatureGate]))
788790
return r.dlRuntime
789791
}
790792

pkg/epp/datalayer/runtime.go

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,20 +59,30 @@ type Runtime struct {
5959

6060
collectors *collectorManager // per-endpoint poller, keyed by namespaced name
6161
logger logr.Logger // Set in Configure; used where no context is available (e.g. ReleaseEndpoint).
62+
63+
multiCluster bool // marks created endpoints as peer EPPs
6264
}
6365

6466
const (
6567
defaultRefreshInterval = 50 * time.Millisecond
6668
)
6769

70+
// RuntimeOption configures a Runtime at construction.
71+
type RuntimeOption func(*Runtime)
72+
73+
// WithMultiCluster marks endpoints the Runtime creates as peer EPPs.
74+
func WithMultiCluster(v bool) RuntimeOption {
75+
return func(r *Runtime) { r.multiCluster = v }
76+
}
77+
6878
// NewRuntime creates a new Runtime with the given polling interval.
6979
// If duration is <= 0, uses the defaultRefreshInterval.
70-
func NewRuntime(pollingInterval time.Duration) *Runtime {
80+
func NewRuntime(pollingInterval time.Duration, opts ...RuntimeOption) *Runtime {
7181
interval := defaultRefreshInterval
7282
if pollingInterval > 0 {
7383
interval = pollingInterval
7484
}
75-
return &Runtime{
85+
r := &Runtime{
7686
pollingInterval: interval,
7787
dispatchers: newPollingDispatchers(),
7888
notification: newNotificationManager(),
@@ -81,6 +91,10 @@ func NewRuntime(pollingInterval time.Duration) *Runtime {
8191
collectors: newCollectorManager(),
8292
logger: logr.Discard(),
8393
}
94+
for _, o := range opts {
95+
o(r)
96+
}
97+
return r
8498
}
8599

86100
// Configure is called to transform the configuration information into the Runtime's
@@ -372,6 +386,9 @@ func (r *Runtime) NewEndpoint(ctx context.Context, endpointMetadata *fwkdl.Endpo
372386
logger = logger.WithValues("endpoint", endpointMetadata.GetNamespacedName())
373387

374388
endpoint := fwkdl.NewEndpoint(endpointMetadata, nil)
389+
if r.multiCluster {
390+
fwkdl.SetMultiCluster(endpoint)
391+
}
375392

376393
dispatchers := make([]fwkdl.PollingDispatcher, 0, r.dispatchers.Count())
377394
for _, d := range r.dispatchers.Dispatchers() {
Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
/*
2+
Copyright 2026 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package datalayer
18+
19+
// MultiClusterFeatureGate enables multi-cluster mode.
20+
const MultiClusterFeatureGate = "multicluster"
21+
22+
// MultiClusterAttrKey marks an endpoint as a peer EPP.
23+
const MultiClusterAttrKey = "llm-d.ai/multicluster"
24+
25+
type multiClusterAttr bool
26+
27+
func (m multiClusterAttr) Clone() Cloneable { return m }
28+
29+
// SetMultiCluster marks ep as a peer EPP.
30+
func SetMultiCluster(ep Endpoint) {
31+
ep.GetAttributes().Put(MultiClusterAttrKey, multiClusterAttr(true))
32+
}
33+
34+
// IsMultiCluster reports whether ep is a peer EPP. Absence is false.
35+
func IsMultiCluster(ep Endpoint) bool {
36+
if v, ok := ep.GetAttributes().Get(MultiClusterAttrKey); ok {
37+
if m, ok := v.(multiClusterAttr); ok {
38+
return bool(m)
39+
}
40+
}
41+
return false
42+
}
Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
Copyright 2026 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package datalayer
18+
19+
import (
20+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
)
24+
25+
func TestMultiClusterAttribute(t *testing.T) {
26+
// Absence reads as not-multi-cluster.
27+
ep := NewEndpoint(nil, nil)
28+
assert.False(t, IsMultiCluster(ep), "unmarked endpoint should not be multi-cluster")
29+
30+
// A marked endpoint reads back true through the attribute map.
31+
SetMultiCluster(ep)
32+
assert.True(t, IsMultiCluster(ep), "marked endpoint should read back as multi-cluster")
33+
}

0 commit comments

Comments
 (0)