Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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 charts/gateway-helm/templates/_rbac.tpl
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@ verbs:
apiGroups:
- gateway.envoyproxy.io
resources:
- envoyproxies/status
- envoypatchpolicies/status
- clienttrafficpolicies/status
- backendtrafficpolicies/status
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,18 @@ envoyProxyForGatewayClass:
"@type": type.googleapis.com/envoy.extensions.resource_monitors.downstream_connections.v3.DownstreamConnectionsConfig
max_active_downstream_connections: 50000
logging: {}
status: {}
status:
ancestors:
- ancestorRef:
group: gateway.networking.k8s.io
kind: GatewayClass
name: eg
conditions:
- lastTransitionTime: null
message: EnvoyProxy has been accepted.
reason: Accepted
status: "True"
type: Accepted
gatewayClass:
apiVersion: gateway.networking.k8s.io/v1
kind: GatewayClass
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,18 @@ envoyProxyForGatewayClass:
logging:
level:
default: warn
status: {}
status:
ancestors:
- ancestorRef:
group: gateway.networking.k8s.io
kind: GatewayClass
name: eg
conditions:
- lastTransitionTime: null
message: dynamic_resources cannot be modified
reason: InvalidParameters
status: "False"
type: Accepted
gatewayClass:
apiVersion: gateway.networking.k8s.io/v1
kind: GatewayClass
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,18 @@ envoyProxyForGatewayClass:
externalTrafficPolicy: Local
type: LoadBalancer
type: Kubernetes
status: {}
status:
ancestors:
- ancestorRef:
group: gateway.networking.k8s.io
kind: GatewayClass
name: eg
conditions:
- lastTransitionTime: null
message: EnvoyProxy has been accepted.
reason: Accepted
status: "True"
type: Accepted
gatewayClass:
apiVersion: gateway.networking.k8s.io/v1
kind: GatewayClass
Expand Down
7 changes: 5 additions & 2 deletions internal/gatewayapi/contexts.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,10 @@ import (
type GatewayContext struct {
*gwapiv1.Gateway

listeners []*ListenerContext
envoyProxy *egv1a1.EnvoyProxy
listeners []*ListenerContext
envoyProxy *egv1a1.EnvoyProxy
envoyProxyFromGateway bool

backendTLS *egv1a1.BackendTLSConfig
}

Expand Down Expand Up @@ -105,6 +107,7 @@ func (g *GatewayContext) attachEnvoyProxy(resources *resource.Resources, epMap m
if string(ref.Group) == egv1a1.GroupVersion.Group && ref.Kind == egv1a1.KindEnvoyProxy {
ep, exists := epMap[types.NamespacedName{Namespace: g.Namespace, Name: ref.Name}]
if exists {
g.envoyProxyFromGateway = true
gatewayProxy = ep
}
}
Expand Down
65 changes: 64 additions & 1 deletion internal/gatewayapi/runner/runner.go
Original file line number Diff line number Diff line change
Expand Up @@ -112,14 +112,33 @@ func mergePolicyStatus(aggregated aggregatedPolicyStatus, incoming *gwapiv1.Poli
return aggregated
}

aggregated.status.Ancestors = append(aggregated.status.Ancestors, incoming.Ancestors...)
// Prevent self-merge when aggregated and incoming reference the same object
if aggregated.status != incoming {
aggregated.status.Ancestors = append(aggregated.status.Ancestors, incoming.Ancestors...)
}
if generation > aggregated.generation {
aggregated.generation = generation
}

return aggregated
}

func mergeEnvoyProxyStatus(aggregated, incoming *egv1a1.EnvoyProxyStatus) *egv1a1.EnvoyProxyStatus {
if incoming == nil {
return aggregated
}

if aggregated == nil {
return incoming
}

// Prevent self-merge when aggregated and incoming reference the same object
if aggregated != incoming {
aggregated.Ancestors = append(aggregated.Ancestors, incoming.Ancestors...)
}
return aggregated
}

func New(cfg *Config) *Runner {
return &Runner{
Config: *cfg,
Expand Down Expand Up @@ -206,6 +225,13 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
// There is only 1 key which is the controller name
// so when a delete is triggered, delete all keys
if update.Delete || valWrapper == nil || valWrapper.Resources == nil {
// Clear EnvoyProxy statuses before deleting to remove stale ancestor conditions
for key := range r.keyCache.EnvoyProxyStatus {
emptyStatus := &egv1a1.EnvoyProxyStatus{
Ancestors: []egv1a1.EnvoyProxyAncestorStatus{},
}
r.ProviderResources.EnvoyProxyStatuses.Store(key, emptyStatus)
}
r.deleteAllKeys()
return
}
Expand All @@ -229,6 +255,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
var tlsRouteStatusCount, tcpRouteStatusCount, udpRouteStatusCount int
var backendTLSPolicyStatusCount, clientTrafficPolicyStatusCount, backendTrafficPolicyStatusCount int
var securityPolicyStatusCount, envoyExtensionPolicyStatusCount, backendStatusCount, extensionServerPolicyStatusCount int
var envoyproxyStatusCount int

// `aggregatedStatuses` aggregates status result of resources from all
// parents/ancestors, and then stores the status once for every resource.
Expand All @@ -244,6 +271,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
SecurityPolicies map[types.NamespacedName]aggregatedPolicyStatus
EnvoyExtensionPolicies map[types.NamespacedName]aggregatedPolicyStatus
ExtensionServerPolicies map[message.NamespacedNameAndGVK]aggregatedPolicyStatus
EnvoyProxies map[types.NamespacedName]*egv1a1.EnvoyProxyStatus
}{
HTTPRoutes: make(map[types.NamespacedName]aggregatedRouteStatus),
GRPCRoutes: make(map[types.NamespacedName]aggregatedRouteStatus),
Expand All @@ -256,6 +284,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
SecurityPolicies: make(map[types.NamespacedName]aggregatedPolicyStatus),
EnvoyExtensionPolicies: make(map[types.NamespacedName]aggregatedPolicyStatus),
ExtensionServerPolicies: make(map[message.NamespacedNameAndGVK]aggregatedPolicyStatus),
EnvoyProxies: make(map[types.NamespacedName]*egv1a1.EnvoyProxyStatus),
}

span.AddEvent("translate", trace.WithAttributes(attribute.Int("resources.count", len(*val))))
Expand Down Expand Up @@ -447,6 +476,15 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
aggregatedStatuses.ExtensionServerPolicies[key] = mergePolicyStatus(aggregatedStatuses.ExtensionServerPolicies[key], &policyStatus, extServerPolicy.GetGeneration())
}
}
// EnvoyProxy status
for _, ep := range result.EnvoyProxiesForGateways {
r.Logger.Info("update envoyproxy status", "key", utils.NamespacedName(ep))
aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)] = mergeEnvoyProxyStatus(aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)], &ep.Status)
}
if ep := result.EnvoyProxyForGatewayClass; ep != nil {
r.Logger.Info("update envoyproxy status", "key", utils.NamespacedName(ep))
aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)] = mergeEnvoyProxyStatus(aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)], &ep.Status)
}
statusUpdateSpan.End()
}

Expand Down Expand Up @@ -533,6 +571,15 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
delete(keysToDelete.ExtensionServerPolicyStatus, key)
r.keyCache.ExtensionServerPolicyStatus[key] = true
}
for key, entry := range aggregatedStatuses.EnvoyProxies {
s := egv1a1.EnvoyProxyStatus{
Ancestors: entry.Ancestors,
}
r.ProviderResources.EnvoyProxyStatuses.Store(key, &s)
envoyproxyStatusCount++
delete(keysToDelete.EnvoyProxyStatus, key)
r.keyCache.EnvoyProxyStatus[key] = true
}
// Publish aggregated metrics
message.PublishMetric(message.Metadata{Runner: r.Name(), Message: message.InfraIRMessageName}, infraIRCount)
message.PublishMetric(message.Metadata{Runner: r.Name(), Message: message.XDSIRMessageName}, xdsIRCount)
Expand All @@ -551,6 +598,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
message.PublishMetric(message.Metadata{Runner: r.Name(), Message: message.EnvoyExtensionPolicyStatusMessageName}, envoyExtensionPolicyStatusCount)
message.PublishMetric(message.Metadata{Runner: r.Name(), Message: message.BackendStatusMessageName}, backendStatusCount)
message.PublishMetric(message.Metadata{Runner: r.Name(), Message: message.ExtensionServerPoliciesStatusMessageName}, extensionServerPolicyStatusCount)
message.PublishMetric(message.Metadata{Runner: r.Name(), Message: message.EnvoyProxyStatusMessageName}, envoyproxyStatusCount)

// Delete keys using mark and sweep
r.deleteKeys(keysToDelete)
Expand Down Expand Up @@ -660,6 +708,9 @@ func (r *Runner) deleteAllKeys() {
for key := range r.keyCache.BackendStatus {
r.ProviderResources.BackendStatuses.Delete(key)
}
for key := range r.keyCache.EnvoyProxyStatus {
r.ProviderResources.EnvoyProxyStatuses.Delete(key)
}

// Clear all tracking
r.keyCache = newKeyCache()
Expand All @@ -684,6 +735,7 @@ type KeyCache struct {
SecurityPolicyStatus map[types.NamespacedName]bool
EnvoyExtensionPolicyStatus map[types.NamespacedName]bool
ExtensionServerPolicyStatus map[message.NamespacedNameAndGVK]bool
EnvoyProxyStatus map[types.NamespacedName]bool

BackendStatus map[types.NamespacedName]bool
}
Expand Down Expand Up @@ -737,6 +789,9 @@ func (kc *KeyCache) copy() *KeyCache {
for key := range kc.ExtensionServerPolicyStatus {
copied.ExtensionServerPolicyStatus[key] = true
}
for key := range kc.EnvoyProxyStatus {
copied.EnvoyProxyStatus[key] = true
}
for key := range kc.BackendStatus {
copied.BackendStatus[key] = true
}
Expand All @@ -760,6 +815,7 @@ func newKeyCache() *KeyCache {
SecurityPolicyStatus: make(map[types.NamespacedName]bool),
EnvoyExtensionPolicyStatus: make(map[types.NamespacedName]bool),
ExtensionServerPolicyStatus: make(map[message.NamespacedNameAndGVK]bool),
EnvoyProxyStatus: make(map[types.NamespacedName]bool),
BackendStatus: make(map[types.NamespacedName]bool),
}
}
Expand Down Expand Up @@ -812,6 +868,9 @@ func (r *Runner) populateKeyCache() {
for key := range r.ProviderResources.ExtensionPolicyStatuses.LoadAll() {
r.keyCache.ExtensionServerPolicyStatus[key] = true
}
for key := range r.ProviderResources.EnvoyProxyStatuses.LoadAll() {
r.keyCache.EnvoyProxyStatus[key] = true
}
for key := range r.ProviderResources.BackendStatuses.LoadAll() {
r.keyCache.BackendStatus[key] = true
}
Expand Down Expand Up @@ -879,6 +938,10 @@ func (r *Runner) deleteKeys(kc *KeyCache) {
r.ProviderResources.ExtensionPolicyStatuses.Delete(key)
delete(r.keyCache.ExtensionServerPolicyStatus, key)
}
for key := range kc.EnvoyProxyStatus {
r.ProviderResources.EnvoyProxyStatuses.Delete(key)
delete(r.keyCache.EnvoyProxyStatus, key)
}
for key := range kc.BackendStatus {
r.ProviderResources.BackendStatuses.Delete(key)
delete(r.keyCache.BackendStatus, key)
Expand Down
88 changes: 88 additions & 0 deletions internal/gatewayapi/runner/runner_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,94 @@ func TestMergePolicyStatus(t *testing.T) {
entry = mergePolicyStatus(entry, &gwapiv1.PolicyStatus{}, 4)
require.Equal(t, int64(9), entry.generation)
})

t.Run("self-merge prevention", func(t *testing.T) {
status := &gwapiv1.PolicyStatus{
Ancestors: []gwapiv1.PolicyAncestorStatus{
{
AncestorRef: gwapiv1.ParentReference{Name: gwapiv1.ObjectName("gw-a")},
ControllerName: gwapiv1a2.GatewayController(controllerName),
},
},
}

// First merge - status becomes both aggregated and incoming
entry := mergePolicyStatus(aggregatedPolicyStatus{}, status, 1)
// Second merge with same status - should not duplicate
entry = mergePolicyStatus(entry, status, 2)

// Should still have only one ancestor, not duplicated
require.Len(t, entry.status.Ancestors, 1)
require.Equal(t, gwapiv1.ObjectName("gw-a"), entry.status.Ancestors[0].AncestorRef.Name)
})
}

func TestMergeEnvoyProxyStatus(t *testing.T) {
t.Run("nil incoming keeps existing entry", func(t *testing.T) {
existing := &egv1a1.EnvoyProxyStatus{
Ancestors: []egv1a1.EnvoyProxyAncestorStatus{
{
AncestorRef: gwapiv1.ParentReference{Name: gwapiv1.ObjectName("gc-a")},
},
},
}

got := mergeEnvoyProxyStatus(existing, nil)
require.Same(t, existing, got)
})

t.Run("nil existing takes incoming", func(t *testing.T) {
incoming := &egv1a1.EnvoyProxyStatus{
Ancestors: []egv1a1.EnvoyProxyAncestorStatus{
{
AncestorRef: gwapiv1.ParentReference{Name: gwapiv1.ObjectName("gc-a")},
},
},
}

got := mergeEnvoyProxyStatus(nil, incoming)
require.Same(t, incoming, got)
})

t.Run("appends ancestors", func(t *testing.T) {
first := &egv1a1.EnvoyProxyStatus{
Ancestors: []egv1a1.EnvoyProxyAncestorStatus{
{
AncestorRef: gwapiv1.ParentReference{Name: gwapiv1.ObjectName("gc-a")},
},
},
}
second := &egv1a1.EnvoyProxyStatus{
Ancestors: []egv1a1.EnvoyProxyAncestorStatus{
{
AncestorRef: gwapiv1.ParentReference{Name: gwapiv1.ObjectName("gw-b")},
},
},
}

got := mergeEnvoyProxyStatus(first, second)

require.Len(t, got.Ancestors, 2)
require.Equal(t, gwapiv1.ObjectName("gc-a"), got.Ancestors[0].AncestorRef.Name)
require.Equal(t, gwapiv1.ObjectName("gw-b"), got.Ancestors[1].AncestorRef.Name)
})

t.Run("self-merge prevention", func(t *testing.T) {
status := &egv1a1.EnvoyProxyStatus{
Ancestors: []egv1a1.EnvoyProxyAncestorStatus{
{
AncestorRef: gwapiv1.ParentReference{Name: gwapiv1.ObjectName("gc-a")},
},
},
}

// Merge status with itself - should not duplicate
got := mergeEnvoyProxyStatus(status, status)

// Should still have only one ancestor, not duplicated
require.Len(t, got.Ancestors, 1)
require.Equal(t, gwapiv1.ObjectName("gc-a"), got.Ancestors[0].AncestorRef.Name)
})
}

func TestMergeRouteStatus(t *testing.T) {
Expand Down
44 changes: 44 additions & 0 deletions internal/gatewayapi/status/envoyproxy.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
// Copyright Envoy Gateway Authors
// SPDX-License-Identifier: Apache-2.0
// The full text of the Apache license is available in the LICENSE file at
// the root of the repo.

package status

import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"

egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
)

func UpdateEnvoyProxyStatusAccepted(ep *egv1a1.EnvoyProxy, ancestor *gwapiv1.ParentReference,
reason egv1a1.EnvoyProxyConditionReason, msg string,
) {
if ep == nil || ancestor == nil {
return
}
status := metav1.ConditionTrue
if reason != egv1a1.EnvoyProxyReasonAccepted {
status = metav1.ConditionFalse
}

cond := newCondition(string(egv1a1.EnvoyProxyConditionAccepted), status,
string(reason), msg, ep.Generation)

for i := range ep.Status.Ancestors {
item := ep.Status.Ancestors[i]
if ancestorRefsEqual(&item.AncestorRef, ancestor) {
ep.Status.Ancestors[i].Conditions = MergeConditions(item.Conditions, cond)
return
}
}

// ancestor not found, append a new one
ep.Status.Ancestors = append(ep.Status.Ancestors, egv1a1.EnvoyProxyAncestorStatus{
AncestorRef: *ancestor,
Conditions: []metav1.Condition{
cond,
},
})
}
Loading