This is an automated email from the ASF dual-hosted git repository. bzp2010 pushed a commit to branch bzp/feat-record-gw-instance-status-accurately in repository https://gitbox.apache.org/repos/asf/apisix-ingress-controller.git
commit 6e2efa1b87d11dc03a11380546f674acc4fa652d Author: bzp2010 <[email protected]> AuthorDate: Fri Sep 4 00:05:27 2026 +0800 feat: create per-endpoint event and link to gateway proxy --- internal/manager/run.go | 1 + internal/provider/apisix/status.go | 43 ++++++++++++++++++++++-- internal/provider/apisix/status_test.go | 59 +++++++++++++++++++++++++++++++-- internal/provider/options.go | 6 ++++ test/e2e/crds/v2/status.go | 13 ++++++++ 5 files changed, 118 insertions(+), 4 deletions(-) diff --git a/internal/manager/run.go b/internal/manager/run.go index b08ae92f..18fd0c75 100644 --- a/internal/manager/run.go +++ b/internal/manager/run.go @@ -223,6 +223,7 @@ func Run(ctx context.Context, logger logr.Logger) error { SyncPeriod: config.ControllerConfig.ProviderConfig.SyncPeriod.Duration, InitSyncDelay: config.ControllerConfig.ProviderConfig.InitSyncDelay.Duration, ListenerPortMatchMode: config.ControllerConfig.ListenerPortMatchMode, + EventRecorder: mgr.GetEventRecorderFor("apisix-provider"), } provider, err := provider.New(providerType, logger, updater.Writer(), readier, providerOptions) if err != nil { diff --git a/internal/provider/apisix/status.go b/internal/provider/apisix/status.go index 7c888a0d..8f12eb70 100644 --- a/internal/provider/apisix/status.go +++ b/internal/provider/apisix/status.go @@ -21,6 +21,7 @@ import ( "fmt" "strings" + corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client" gatewayv1 "sigs.k8s.io/gateway-api/apis/v1" @@ -61,6 +62,9 @@ func (d *apisixProvider) handleStatusUpdate(statusUpdateMap map[types.Namespaced for nnk := range d.statusUpdateMap { if _, ok := statusUpdateMap[nnk]; !ok { d.updateStatus(nnk, successCondition(nnk)) + if nnk.Kind == types.KindGatewayProxy { + d.recordGatewayProxyRecoveredEvent(nnk) + } } } // Update the failure history with the current failure set. @@ -98,6 +102,38 @@ func newGatewayProxyDataPlaneAvailableCondition(available bool, reason, msg stri } } +// recordGatewayProxyRecoveredEvent fires once a GatewayProxy leaves the failure +// history entirely, i.e. every instance took the last sync. +func (d *apisixProvider) recordGatewayProxyRecoveredEvent(nnk types.NamespacedNameKind) { + if d.EventRecorder == nil { + return + } + gatewayProxy := &apiv1alpha1.GatewayProxy{ + ObjectMeta: metav1.ObjectMeta{Name: nnk.Name, Namespace: nnk.Namespace}, + } + d.EventRecorder.Event(gatewayProxy, corev1.EventTypeNormal, GatewayProxyReasonDataPlaneAvailable, + "every data plane instance took the last sync") +} + +// recordFailedEndpointEvents fires one Warning event per failed EndpointStatus entry, +// so each instance's own failure history (when it started, how often) is visible on +// its own, not folded into everyone else's. +func (d *apisixProvider) recordFailedEndpointEvents(nnk types.NamespacedNameKind, endpoints []adctypes.EndpointStatus) { + if d.EventRecorder == nil { + return + } + gatewayProxy := &apiv1alpha1.GatewayProxy{ + ObjectMeta: metav1.ObjectMeta{Name: nnk.Name, Namespace: nnk.Namespace}, + } + for _, ep := range endpoints { + if ep.Success { + continue + } + d.EventRecorder.Event(gatewayProxy, corev1.EventTypeWarning, GatewayProxyReasonDataPlaneInstanceUnavailable, + fmt.Sprintf("%s: %s", ep.Server, ep.Reason)) + } +} + //nolint:gocyclo func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition metav1.Condition) { switch nnk.Kind { @@ -330,7 +366,7 @@ func (d *apisixProvider) handleEmptyFailedStatuses( statusUpdateMap map[types.NamespacedNameKind][]string, ) { if msg := unavailableEndpointsMessage(failedStatus.EndpointStatuses); msg != "" { - d.markGatewayProxyDataPlaneUnavailable(configName, msg, statusUpdateMap) + d.markGatewayProxyDataPlaneUnavailable(configName, msg, failedStatus.EndpointStatuses, statusUpdateMap) return } d.smearAllResources(configName, failedStatus.Error(), statusUpdateMap) @@ -353,10 +389,12 @@ func unavailableEndpointsMessage(endpoints []adctypes.EndpointStatus) string { len(failed), len(endpoints), strings.Join(failed, "; ")) } -// markGatewayProxyDataPlaneUnavailable marks the GatewayProxy configName names. +// markGatewayProxyDataPlaneUnavailable marks the GatewayProxy configName names, and +// fires one event per failed endpoint. func (d *apisixProvider) markGatewayProxyDataPlaneUnavailable( configName string, msg string, + endpoints []adctypes.EndpointStatus, statusUpdateMap map[types.NamespacedNameKind][]string, ) { var gatewayProxy types.NamespacedNameKind @@ -365,6 +403,7 @@ func (d *apisixProvider) markGatewayProxyDataPlaneUnavailable( return } statusUpdateMap[gatewayProxy] = append(statusUpdateMap[gatewayProxy], msg) + d.recordFailedEndpointEvents(gatewayProxy, endpoints) } // smearAllResources is the last resort when nothing can be attributed: it marks every diff --git a/internal/provider/apisix/status_test.go b/internal/provider/apisix/status_test.go index d4c75fe2..79198fdc 100644 --- a/internal/provider/apisix/status_test.go +++ b/internal/provider/apisix/status_test.go @@ -18,10 +18,12 @@ package apisix import ( + "strings" "testing" "github.com/go-logr/logr" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/record" adctypes "github.com/apache/apisix-ingress-controller/api/adc" apiv2 "github.com/apache/apisix-ingress-controller/api/v2" @@ -112,7 +114,7 @@ func TestMarkGatewayProxyDataPlaneUnavailableTargetsTheConfigNameItself(t *testi d := &apisixProvider{log: logr.Discard()} statusUpdateMap := map[types.NamespacedNameKind][]string{} - d.markGatewayProxyDataPlaneUnavailable("GatewayProxy/ns/gp", "boom", statusUpdateMap) + d.markGatewayProxyDataPlaneUnavailable("GatewayProxy/ns/gp", "boom", nil, statusUpdateMap) want := types.NamespacedNameKind{Kind: types.KindGatewayProxy, Namespace: "ns", Name: "gp"} if got := statusUpdateMap[want]; len(got) != 1 || got[0] != "boom" { @@ -127,9 +129,62 @@ func TestMarkGatewayProxyDataPlaneUnavailableIgnoresAnUnparseableConfigName(t *t d := &apisixProvider{log: logr.Discard()} statusUpdateMap := map[types.NamespacedNameKind][]string{} - d.markGatewayProxyDataPlaneUnavailable("not-a-valid-key", "boom", statusUpdateMap) + d.markGatewayProxyDataPlaneUnavailable("not-a-valid-key", "boom", nil, statusUpdateMap) if len(statusUpdateMap) != 0 { t.Errorf("expected nothing marked for an unparseable configName, got %v", statusUpdateMap) } } + +func TestRecordFailedEndpointEventsFiresOneWarningPerFailedEndpoint(t *testing.T) { + recorder := record.NewFakeRecorder(2) + d := &apisixProvider{log: logr.Discard()} + d.EventRecorder = recorder + nnk := types.NamespacedNameKind{Kind: types.KindGatewayProxy, Namespace: "ns", Name: "gp"} + + d.recordFailedEndpointEvents(nnk, []adctypes.EndpointStatus{ + {Server: "http://apisix-1:9180";, Success: true}, + {Server: "http://apisix-2:9180";, Success: false, Reason: "connection refused"}, + {Server: "http://apisix-3:9180";, Success: false, Reason: "TLS handshake failed"}, + }) + + first := <-recorder.Events + if !strings.Contains(first, "Warning") || !strings.Contains(first, "DataPlaneInstanceUnavailable") || + !strings.Contains(first, "http://apisix-2:9180: connection refused") { + t.Errorf("first event = %q, want the apisix-2 failure", first) + } + + second := <-recorder.Events + if !strings.Contains(second, "Warning") || !strings.Contains(second, "DataPlaneInstanceUnavailable") || + !strings.Contains(second, "http://apisix-3:9180: TLS handshake failed") { + t.Errorf("second event = %q, want the apisix-3 failure", second) + } + + select { + case e := <-recorder.Events: + t.Errorf("unexpected third event %q, the successful endpoint should not fire one", e) + default: + } +} + +func TestRecordFailedEndpointEventsNoopsWithoutARecorder(t *testing.T) { + d := &apisixProvider{log: logr.Discard()} + nnk := types.NamespacedNameKind{Kind: types.KindGatewayProxy, Namespace: "ns", Name: "gp"} + + // Must not panic when no EventRecorder was configured. + d.recordFailedEndpointEvents(nnk, []adctypes.EndpointStatus{{Server: "http://apisix-1:9180";, Success: false}}) +} + +func TestRecordGatewayProxyRecoveredEventFiresNormal(t *testing.T) { + recorder := record.NewFakeRecorder(1) + d := &apisixProvider{log: logr.Discard()} + d.EventRecorder = recorder + nnk := types.NamespacedNameKind{Kind: types.KindGatewayProxy, Namespace: "ns", Name: "gp"} + + d.recordGatewayProxyRecoveredEvent(nnk) + + event := <-recorder.Events + if !strings.Contains(event, "Normal") || !strings.Contains(event, "DataPlaneAvailable") { + t.Errorf("recovery event = %q, want it to contain Normal and DataPlaneAvailable", event) + } +} diff --git a/internal/provider/options.go b/internal/provider/options.go index c47e7ce9..7465f520 100644 --- a/internal/provider/options.go +++ b/internal/provider/options.go @@ -20,6 +20,8 @@ package provider import ( "time" + "k8s.io/client-go/tools/record" + "github.com/apache/apisix-ingress-controller/internal/controller/config" ) @@ -34,6 +36,7 @@ type Options struct { DefaultBackendMode string DefaultResolveEndpoints bool ListenerPortMatchMode config.ListenerPortMatchMode + EventRecorder record.EventRecorder } func (o *Options) ApplyToList(lo *Options) { @@ -55,6 +58,9 @@ func (o *Options) ApplyToList(lo *Options) { if o.ListenerPortMatchMode != "" { lo.ListenerPortMatchMode = o.ListenerPortMatchMode } + if o.EventRecorder != nil { + lo.EventRecorder = o.EventRecorder + } } func (o *Options) ApplyOptions(opts []Option) *Options { diff --git a/test/e2e/crds/v2/status.go b/test/e2e/crds/v2/status.go index 54bd4c06..3c06a0c2 100644 --- a/test/e2e/crds/v2/status.go +++ b/test/e2e/crds/v2/status.go @@ -291,6 +291,19 @@ spec: ), ) + By("check a Warning event was recorded on the GatewayProxy") + s.RetryAssertion(func() string { + output, _ := s.GetOutputFromString("events", + "--field-selector", "involvedObject.kind=GatewayProxy,involvedObject.name="+name, + "-n", s.Namespace()) + return output + }).Should( + And( + ContainSubstring("Warning"), + ContainSubstring("DataPlaneInstanceUnavailable"), + ), + ) + By("check the ApisixRoute itself is not marked as failed") s.RetryAssertion(func() string { output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml", "-n", s.Namespace())
