This is an automated email from the ASF dual-hosted git repository.
bzp2010 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix-ingress-controller.git
The following commit(s) were added to refs/heads/master by this push:
new f14782d3 feat: report dp instance unavailable to gateway proxy (#2856)
f14782d3 is described below
commit f14782d39f1a922241d698319ad67420cd4b2393
Author: Zeping Bai <[email protected]>
AuthorDate: Mon Sep 14 14:03:48 2026 +0800
feat: report dp instance unavailable to gateway proxy (#2856)
---
.github/workflows/apisix-e2e-test.yml | 2 +-
api/v1alpha1/gatewayproxy_types.go | 14 +
api/v1alpha1/zz_generated.deepcopy.go | 23 ++
.../bases/apisix.apache.org_gatewayproxies.yaml | 71 ++++
config/manager/kustomization.yaml | 2 +-
config/rbac/role.yaml | 1 +
docs/en/latest/reference/api-reference.md | 2 +
docs/en/latest/reference/troubleshoot.md | 34 ++
internal/adc/client/client.go | 2 +-
internal/adc/client/executor.go | 12 +-
internal/adc/client/executor_test.go | 40 ++
internal/controller/status/updater.go | 6 +
internal/controller/status/updater_test.go | 146 +++++++
internal/manager/controllers.go | 1 +
internal/manager/run.go | 2 +
internal/provider/apisix/provider.go | 70 ++--
internal/provider/apisix/status.go | 328 ++++++++++------
internal/provider/apisix/status_test.go | 437 +++++++++++++++++++++
internal/provider/options.go | 14 +
test/e2e/crds/v2/route.go | 19 +-
test/e2e/crds/v2/status.go | 130 +++++-
test/e2e/framework/manifests/ingress.yaml | 3 +-
test/e2e/gatewayapi/status.go | 11 +-
23 files changed, 1201 insertions(+), 169 deletions(-)
diff --git a/.github/workflows/apisix-e2e-test.yml
b/.github/workflows/apisix-e2e-test.yml
index 5c7faf60..1a00f63b 100644
--- a/.github/workflows/apisix-e2e-test.yml
+++ b/.github/workflows/apisix-e2e-test.yml
@@ -30,7 +30,7 @@ concurrency:
cancel-in-progress: true
env:
- ADC_RUST_VERSION: "0.30.2"
+ ADC_RUST_VERSION: "0.30.3"
jobs:
e2e-test:
diff --git a/api/v1alpha1/gatewayproxy_types.go
b/api/v1alpha1/gatewayproxy_types.go
index f9a5aa8e..ee5d5621 100644
--- a/api/v1alpha1/gatewayproxy_types.go
+++ b/api/v1alpha1/gatewayproxy_types.go
@@ -170,6 +170,7 @@ type ProviderService struct {
}
// +kubebuilder:object:root=true
+// +kubebuilder:subresource:status
// GatewayProxy defines configuration for the gateway proxy instances used to
route traffic to services.
type GatewayProxy struct {
metav1.TypeMeta `json:",inline"`
@@ -178,6 +179,19 @@ type GatewayProxy struct {
// GatewayProxySpec defines configuration of gateway proxy instances,
// including networking settings, global plugins, and plugin metadata.
Spec GatewayProxySpec `json:"spec,omitempty"`
+
+ // Status defines the current state of Gateway Proxy.
+ //
+ // +kubebuilder:default={conditions: {{type: "DataPlaneAvailable",
status: "Unknown", reason:"Pending", message:"Waiting for controller",
lastTransitionTime: "1970-01-01T00:00:00Z"}}}
+ // +optional
+ Status GatewayProxyStatus `json:"status,omitempty"`
+}
+
+// GatewayProxyStatus defines the observed state of GatewayProxy.
+type GatewayProxyStatus struct {
+ // Conditions describe the current state of the data plane instances
this GatewayProxy addresses.
+ // +optional
+ Conditions []metav1.Condition `json:"conditions,omitempty"`
}
// +kubebuilder:object:root=true
diff --git a/api/v1alpha1/zz_generated.deepcopy.go
b/api/v1alpha1/zz_generated.deepcopy.go
index 8b9c5438..484e061d 100644
--- a/api/v1alpha1/zz_generated.deepcopy.go
+++ b/api/v1alpha1/zz_generated.deepcopy.go
@@ -462,6 +462,7 @@ func (in *GatewayProxy) DeepCopyInto(out *GatewayProxy) {
out.TypeMeta = in.TypeMeta
in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
in.Spec.DeepCopyInto(&out.Spec)
+ in.Status.DeepCopyInto(&out.Status)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver,
creating a new GatewayProxy.
@@ -589,6 +590,28 @@ func (in *GatewayProxySpec) DeepCopy() *GatewayProxySpec {
return out
}
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver,
writing into out. in must be non-nil.
+func (in *GatewayProxyStatus) DeepCopyInto(out *GatewayProxyStatus) {
+ *out = *in
+ if in.Conditions != nil {
+ in, out := &in.Conditions, &out.Conditions
+ *out = make([]metav1.Condition, len(*in))
+ for i := range *in {
+ (*in)[i].DeepCopyInto(&(*out)[i])
+ }
+ }
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver,
creating a new GatewayProxyStatus.
+func (in *GatewayProxyStatus) DeepCopy() *GatewayProxyStatus {
+ if in == nil {
+ return nil
+ }
+ out := new(GatewayProxyStatus)
+ in.DeepCopyInto(out)
+ return out
+}
+
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver,
writing into out. in must be non-nil.
func (in *GatewayRef) DeepCopyInto(out *GatewayRef) {
*out = *in
diff --git a/config/crd/bases/apisix.apache.org_gatewayproxies.yaml
b/config/crd/bases/apisix.apache.org_gatewayproxies.yaml
index b2865255..0efae1cf 100644
--- a/config/crd/bases/apisix.apache.org_gatewayproxies.yaml
+++ b/config/crd/bases/apisix.apache.org_gatewayproxies.yaml
@@ -203,6 +203,77 @@ spec:
required:
- provider
type: object
+ status:
+ default:
+ conditions:
+ - lastTransitionTime: "1970-01-01T00:00:00Z"
+ message: Waiting for controller
+ reason: Pending
+ status: Unknown
+ type: DataPlaneAvailable
+ description: Status defines the current state of Gateway Proxy.
+ properties:
+ conditions:
+ description: Conditions describe the current state of the data
plane
+ instances this GatewayProxy addresses.
+ items:
+ description: Condition contains details for one aspect of
the current
+ state of this API Resource.
+ properties:
+ lastTransitionTime:
+ description: |-
+ lastTransitionTime is the last time the condition
transitioned from one status to another.
+ This should be when the underlying condition changed.
If that is not known, then using the time when the API field changed is
acceptable.
+ format: date-time
+ type: string
+ message:
+ description: |-
+ message is a human readable message indicating details
about the transition.
+ This may be an empty string.
+ maxLength: 32768
+ type: string
+ observedGeneration:
+ description: |-
+ observedGeneration represents the .metadata.generation
that the condition was set based upon.
+ For instance, if .metadata.generation is currently 12,
but the .status.conditions[x].observedGeneration is 9, the condition is out of
date
+ with respect to the current state of the instance.
+ format: int64
+ minimum: 0
+ type: integer
+ reason:
+ description: |-
+ reason contains a programmatic identifier indicating
the reason for the condition's last transition.
+ Producers of specific condition types may define
expected values and meanings for this field,
+ and whether the values are considered a guaranteed API.
+ The value should be a CamelCase string.
+ This field may not be empty.
+ maxLength: 1024
+ minLength: 1
+ pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
+ type: string
+ status:
+ description: status of the condition, one of True,
False, Unknown.
+ enum:
+ - "True"
+ - "False"
+ - Unknown
+ type: string
+ type:
+ description: type of condition in CamelCase or in
foo.example.com/CamelCase.
+ maxLength: 316
+ pattern:
^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$
+ type: string
+ required:
+ - lastTransitionTime
+ - message
+ - reason
+ - status
+ - type
+ type: object
+ type: array
+ type: object
type: object
served: true
storage: true
+ subresources:
+ status: {}
diff --git a/config/manager/kustomization.yaml
b/config/manager/kustomization.yaml
index 12daa524..7410dcd1 100644
--- a/config/manager/kustomization.yaml
+++ b/config/manager/kustomization.yaml
@@ -17,4 +17,4 @@ images:
newTag: dev
- name: sidecar
newName: ghcr.io/api7/adc
- newTag: 0.29.0
+ newTag: 0.30.3
diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml
index 86b93cc3..b8432ff2 100644
--- a/config/rbac/role.yaml
+++ b/config/rbac/role.yaml
@@ -53,6 +53,7 @@ rules:
- apisixupstreams/status
- backendtrafficpolicies/status
- consumers/status
+ - gatewayproxies/status
- httproutepolicies/status
- l4routepolicies/status
verbs:
diff --git a/docs/en/latest/reference/api-reference.md
b/docs/en/latest/reference/api-reference.md
index 84cf404b..a120c1c6 100644
--- a/docs/en/latest/reference/api-reference.md
+++ b/docs/en/latest/reference/api-reference.md
@@ -407,6 +407,8 @@ GatewayProxySpec defines the desired state of GatewayProxy.
_Appears in:_
- [GatewayProxy](#gatewayproxy)
+
+
#### GatewayRef
diff --git a/docs/en/latest/reference/troubleshoot.md
b/docs/en/latest/reference/troubleshoot.md
index 837ccbe2..26915ada 100644
--- a/docs/en/latest/reference/troubleshoot.md
+++ b/docs/en/latest/reference/troubleshoot.md
@@ -55,6 +55,40 @@ curl "http://127.0.0.1:9180/apisix/admin/routes"; -H
"X-API-KEY: ${ADMIN_API_KEY}
For reference, see [Admin
API](https://apisix.apache.org/docs/apisix/admin-api/).
+## Check Data Plane Instance Availability
+
+When running APISIX in standalone mode with more than one instance, a route
can be reachable through some instances but not others, for example if one
instance is unreachable or rejects the synchronized configuration. This is not
reflected on the affected route's own status, since the route itself was valid;
it shows up on the `GatewayProxy` that addresses those instances.
+
+Check the `DataPlaneAvailable` condition:
+
+```shell
+kubectl get gatewayproxy <gateway-proxy-name> -o yaml
+```
+
+```yaml
+status:
+ conditions:
+ - type: DataPlaneAvailable
+ status: "False"
+ reason: DataPlaneInstanceUnavailable
+ message: "1/3 gateway instance(s) failed to apply the last sync:
http://apisix-2:9180: connection refused"
+```
+
+For the history of which specific instance failed and when, check the
GatewayProxy's events:
+
+```shell
+kubectl describe gatewayproxy <gateway-proxy-name>
+```
+
+```text
+Events:
+ Type Reason Age From Message
+ ---- ------ ---- ---- -------
+ Warning DataPlaneInstanceUnavailable 8s apisix-provider
http://apisix-2:9180: connection refused
+```
+
+Each unreachable or rejecting instance is reported as its own `Warning` event,
so instances failing for different reasons, or at different times, don't get
folded into one message.
+
## Gateway API Routes Return 404
Gateway API HTTPRoute or GRPCRoute resources may return `404` when the Gateway
listener ports do not match the ports that APISIX actually listens on.
diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go
index ae86aae2..76fec1bb 100644
--- a/internal/adc/client/client.go
+++ b/internal/adc/client/client.go
@@ -134,7 +134,7 @@ func (c *Client) Validate(ctx context.Context, task Task)
error {
// SyncInput is one GatewayProxy's complete sync unit. AIC builds it entirely
from its own
// bookkeeping (which resources target this config, their merged translated
snapshot)
-// before handing it over -- this package never reaches back into AIC's state
to gather
+// before handing it over: this package never reaches back into AIC's state to
gather
// anything itself, it only translates, sends, and interprets the response.
type SyncInput struct {
// Name is the cacheKey: the GatewayProxy's own identity.
diff --git a/internal/adc/client/executor.go b/internal/adc/client/executor.go
index a6c6f057..ebe0989c 100644
--- a/internal/adc/client/executor.go
+++ b/internal/adc/client/executor.go
@@ -83,8 +83,8 @@ type ADCServerOpts struct {
CaCert string `json:"caCert,omitempty"`
CacheKey string `json:"cacheKey"`
// BypassCache is only accepted by the /sync task of ADC >= 0.27.0.
Both ADC task
- // schemas reject unknown fields, so omitempty is what keeps every
other request --
- // /validate, and every sync that is not recovering from a rejection --
byte for byte
+ // schemas reject unknown fields, so omitempty is what keeps every
other request,
+ // /validate, and every sync that is not recovering from a rejection,
byte for byte
// what an older ADC server already accepts.
BypassCache bool `json:"bypassCache,omitempty"`
}
@@ -325,7 +325,7 @@ func (e *HTTPADCExecutor) buildHTTPRequest(ctx
context.Context, serverAddr strin
}
// distinctReasons joins every distinct, non-empty reason in failed, in the
order first
-// seen -- several resources failing for the exact same reason (a rejected
conf_version,
+// seen: several resources failing for the exact same reason (a rejected
conf_version,
// say) still reports it once.
func distinctReasons(failed []adctypes.SyncStatus) string {
seen := make(map[string]bool, len(failed))
@@ -360,11 +360,11 @@ func (e *HTTPADCExecutor) handleHTTPResponse(resp
*http.Response, serverAddr str
// - 400: malformed request, unrelated to any backend's content
rejection (see 422).
// - 413: request body over the fixed 100 MB limit.
// - 422: apisix-standalone, every server ended up success:false on
the write itself
- // (rejected the content, unreachable, or a mix -- concurrent
writes aren't
+ // (rejected the content, unreachable, or a mix; concurrent writes
aren't
// cancelled on the first failure).
// - 500: ADC Server failed before attempting the write at all (e.g.
can't reach any
- // server to fetch the current remote state) -- never a verdict on
the write.
- // Only 200/202/422 carry a SyncResult body -- the other three
shouldn't be parsed as
+ // server to fetch the current remote state), never a verdict on
the write.
+ // Only 200/202/422 carry a SyncResult body; the other three shouldn't
be parsed as
// one: most fields are optional, so an unrelated shape can unmarshal
as an empty,
// unremarkable "success".
switch resp.StatusCode {
diff --git a/internal/adc/client/executor_test.go
b/internal/adc/client/executor_test.go
index 12b0cfb4..d7016bb9 100644
--- a/internal/adc/client/executor_test.go
+++ b/internal/adc/client/executor_test.go
@@ -143,6 +143,46 @@ func TestHandleHTTPResponseParsesStructuredReasonOn422(t
*testing.T) {
assert.Equal(t, "unknown plugin foo", addrErr.FailedStatuses[0].Reason)
}
+func TestHandleHTTPResponseParsesTheEventEnvelopeInCamelCase(t *testing.T) {
+ // A literal wire body, not json.Marshal(SyncStatus{...}): marshaling a
Go value and
+ // unmarshaling it right back can't catch a tag mismatch, since both
sides would agree
+ // on whatever the tag currently says. This is what ADC actually sends
(captured from a
+ // real /sync response):
Event.ResourceType/ResourceID/ResourceName/ParentID must
+ // deserialize from resourceType/resourceId/resourceName/parentId, not
+ // resource_type/resource_id/resource_name/parent_id, or
classifySyncResult can never
+ // attribute a failure to the resource it names.
+ e := &HTTPADCExecutor{log: logr.Discard()}
+ body := `{
+ "status": "partial_failure",
+ "total_resources": 2,
+ "success_count": 1,
+ "failed_count": 1,
+ "success": [],
+ "failed": [{
+ "event": {
+ "resourceType": "route",
+ "type": "create",
+ "parentId": "9bf5441c",
+ "resourceId": "43428b9e",
+ "resourceName": "ns_default_rule0"
+ },
+ "reason": "unknown plugin [non-existent-plugin]"
+ }]
+ }`
+
+ err := e.handleHTTPResponse(httpResponse(http.StatusAccepted, body),
"http://apisix:9180";)
+
+ var addrErr types.ADCExecutionServerAddrError
+ require.ErrorAs(t, err, &addrErr)
+ require.Len(t, addrErr.FailedStatuses, 1)
+ event := addrErr.FailedStatuses[0].Event
+ assert.Equal(t, "route", event.ResourceType)
+ assert.Equal(t, "create", event.Type)
+ assert.Equal(t, "9bf5441c", event.ParentID)
+ assert.Equal(t, "43428b9e", event.ResourceID)
+ assert.Equal(t, "ns_default_rule0", event.ResourceName)
+}
+
func
TestHandleHTTPResponseErrorsOnAnUnparseableBodyForAStatusThatShouldCarryOne(t
*testing.T) {
// Nothing actually sends a 2xx/422 body that isn't a SyncResult, but
the parse must
// still fail loudly rather than silently reading as success.
diff --git a/internal/controller/status/updater.go
b/internal/controller/status/updater.go
index ae1e1c28..3d07fd72 100644
--- a/internal/controller/status/updater.go
+++ b/internal/controller/status/updater.go
@@ -254,6 +254,12 @@ func statusEqual(a, b any, opts ...cmp.Option) bool {
return false
}
statusA, statusB = a.Status, b.Status
+ case *v1alpha1.GatewayProxy:
+ b, ok := b.(*v1alpha1.GatewayProxy)
+ if !ok {
+ return false
+ }
+ statusA, statusB = a.Status, b.Status
default:
return false
}
diff --git a/internal/controller/status/updater_test.go
b/internal/controller/status/updater_test.go
new file mode 100644
index 00000000..0d34442c
--- /dev/null
+++ b/internal/controller/status/updater_test.go
@@ -0,0 +1,146 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package status
+
+import (
+ "context"
+ "testing"
+
+ "github.com/go-logr/logr"
+ "github.com/stretchr/testify/require"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/runtime"
+ k8stypes "k8s.io/apimachinery/pkg/types"
+ clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+ "sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+ "github.com/apache/apisix-ingress-controller/api/v1alpha1"
+)
+
+func gatewayProxyWithCondition(condition metav1.Condition)
*v1alpha1.GatewayProxy {
+ return &v1alpha1.GatewayProxy{
+ Status: v1alpha1.GatewayProxyStatus{
+ Conditions: []metav1.Condition{condition},
+ },
+ }
+}
+
+func TestStatusEqualHasAGatewayProxyCase(t *testing.T) {
+ condition := metav1.Condition{
+ Type: "DataPlaneAvailable",
+ Status: metav1.ConditionTrue,
+ Reason: "DataPlaneAvailable",
+ LastTransitionTime: metav1.Now(),
+ }
+ a := gatewayProxyWithCondition(condition)
+ b := gatewayProxyWithCondition(condition)
+
+ if !statusEqual(a, b) {
+ t.Fatal("two GatewayProxy objects with an identical condition
must compare equal, not fall through to the default case")
+ }
+}
+
+func TestStatusEqualIgnoresLastTransitionTimeForGatewayProxy(t *testing.T) {
+ a := gatewayProxyWithCondition(metav1.Condition{
+ Type: "DataPlaneAvailable",
+ Status: metav1.ConditionTrue,
+ Reason: "DataPlaneAvailable",
+ LastTransitionTime: metav1.NewTime(metav1.Now().Add(-1)),
+ })
+ b := gatewayProxyWithCondition(metav1.Condition{
+ Type: "DataPlaneAvailable",
+ Status: metav1.ConditionTrue,
+ Reason: "DataPlaneAvailable",
+ LastTransitionTime: metav1.Now(),
+ })
+
+ if !statusEqual(a, b, cmpIgnoreLastTT) {
+ t.Fatal("a fresh LastTransitionTime alone must not make an
otherwise-unchanged GatewayProxy condition compare unequal")
+ }
+}
+
+func TestStatusEqualDetectsAChangedGatewayProxyCondition(t *testing.T) {
+ a := gatewayProxyWithCondition(metav1.Condition{Type:
"DataPlaneAvailable", Status: metav1.ConditionTrue, Reason:
"DataPlaneAvailable"})
+ b := gatewayProxyWithCondition(metav1.Condition{Type:
"DataPlaneAvailable", Status: metav1.ConditionFalse, Reason:
"DataPlaneInstanceUnavailable"})
+
+ if statusEqual(a, b, cmpIgnoreLastTT) {
+ t.Fatal("a real condition change must still compare unequal")
+ }
+}
+
+// reapplySameGatewayProxyCondition mutates a fresh GatewayProxy fixture
through
+// UpdateHandler.updateStatus twice with a mutator that always writes the same
+// condition values (only LastTransitionTime differs between calls, as a real
+// mutator's `metav1.Now()` would) and returns the object's resourceVersion
after
+// each call.
+func reapplySameGatewayProxyCondition(t *testing.T) (first, second string) {
+ t.Helper()
+
+ scheme := runtime.NewScheme()
+ require.NoError(t, clientgoscheme.AddToScheme(scheme))
+ require.NoError(t, v1alpha1.AddToScheme(scheme))
+
+ gatewayProxy := &v1alpha1.GatewayProxy{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name:
"proxy"},
+ }
+ cli := fake.NewClientBuilder().WithScheme(scheme).
+ WithObjects(gatewayProxy).
+ WithStatusSubresource(gatewayProxy).
+ Build()
+
+ u := NewStatusUpdateHandler(logr.Discard(), cli)
+ mutate := func(obj client.Object) client.Object {
+ cp := obj.(*v1alpha1.GatewayProxy).DeepCopy()
+ cp.Status.Conditions = []metav1.Condition{{
+ Type: "DataPlaneAvailable",
+ Status: metav1.ConditionFalse,
+ Reason: "DataPlaneInstanceUnavailable",
+ Message: "unreachable",
+ LastTransitionTime: metav1.Now(),
+ }}
+ return cp
+ }
+ nnk := k8stypes.NamespacedName{Namespace: "default", Name: "proxy"}
+ ctx := context.Background()
+
+ require.NoError(t, u.updateStatus(ctx, Update{
+ NamespacedName: nnk,
+ Resource: &v1alpha1.GatewayProxy{},
+ Mutator: MutatorFunc(mutate),
+ }))
+ var afterFirst v1alpha1.GatewayProxy
+ require.NoError(t, cli.Get(ctx, nnk, &afterFirst))
+
+ require.NoError(t, u.updateStatus(ctx, Update{
+ NamespacedName: nnk,
+ Resource: &v1alpha1.GatewayProxy{},
+ Mutator: MutatorFunc(mutate),
+ }))
+ var afterSecond v1alpha1.GatewayProxy
+ require.NoError(t, cli.Get(ctx, nnk, &afterSecond))
+
+ return afterFirst.ResourceVersion, afterSecond.ResourceVersion
+}
+
+func TestUpdateStatusSkipsTheWriteWhenTheConditionDidNotActuallyChange(t
*testing.T) {
+ first, second := reapplySameGatewayProxyCondition(t)
+ if first != second {
+ t.Fatalf("reapplying an unchanged condition must not write
again: resourceVersion went from %q to %q", first, second)
+ }
+}
diff --git a/internal/manager/controllers.go b/internal/manager/controllers.go
index 51ec3f91..5cc55435 100644
--- a/internal/manager/controllers.go
+++ b/internal/manager/controllers.go
@@ -71,6 +71,7 @@ import (
// CustomResourceDefinition
//
+kubebuilder:rbac:groups=apisix.apache.org,resources=pluginconfigs,verbs=get;list;watch
//
+kubebuilder:rbac:groups=apisix.apache.org,resources=gatewayproxies,verbs=get;list;watch
+//
+kubebuilder:rbac:groups=apisix.apache.org,resources=gatewayproxies/status,verbs=get;update
//
+kubebuilder:rbac:groups=apisix.apache.org,resources=consumers,verbs=get;list;watch
//
+kubebuilder:rbac:groups=apisix.apache.org,resources=consumers/status,verbs=get;update
//
+kubebuilder:rbac:groups=apisix.apache.org,resources=backendtrafficpolicies,verbs=get;list;watch
diff --git a/internal/manager/run.go b/internal/manager/run.go
index b08ae92f..ed207457 100644
--- a/internal/manager/run.go
+++ b/internal/manager/run.go
@@ -223,6 +223,8 @@ 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"), //nolint:staticcheck
+ K8sClient: mgr.GetClient(),
}
provider, err := provider.New(providerType, logger, updater.Writer(),
readier, providerOptions)
if err != nil {
diff --git a/internal/provider/apisix/provider.go
b/internal/provider/apisix/provider.go
index 3b8139b9..f83df0c0 100644
--- a/internal/provider/apisix/provider.go
+++ b/internal/provider/apisix/provider.go
@@ -79,8 +79,11 @@ type apisixProvider struct {
// record of which cacheKeys it has rebuilt. Unused for every other
backend type.
standaloneSyncer *adcclient.StandaloneSyncer
- updater status.Updater
- statusUpdateMap map[types.NamespacedNameKind][]string
+ updater status.Updater
+ // resourceFailures holds which non-GatewayProxy resources currently
have a sync
+ // error recorded, so the next round that stops seeing one can clear
it. GatewayProxy
+ // keeps no such history: see updateStatusFromSyncResults.
+ resourceFailures map[types.NamespacedNameKind][]string
readier readiness.ReadinessManager
@@ -268,7 +271,7 @@ func (d *apisixProvider) Delete(ctx context.Context, obj
client.Object) error {
}
// applyResourceState upserts a resource's config associations and its
contribution to each
-// target config's cached resource snapshot -- the AIC-side bookkeeping the
adc client
+// target config's cached resource snapshot, the AIC-side bookkeeping the adc
client
// package no longer holds itself.
func (d *apisixProvider) applyResourceState(
rk types.NamespacedNameKind,
@@ -327,27 +330,32 @@ func (d *apisixProvider) evictFromStore(
}
// syncConfigNow reads name's current data (via build, called only once this
cacheKey's
-// lock is actually held) and pushes it -- one atomic read-then-push step per
cacheKey, so
+// lock is actually held) and pushes it, one atomic read-then-push step per
cacheKey, so
// whichever caller is granted the lock decides what to push only once it
holds it: nothing
// it sends can already be stale relative to whatever the other caller
committed to the
// store before losing the race for the same key. See keyedMutex.
+//
+// A nil result means build itself failed before anything could be dispatched
to the data
+// plane: this round never actually reached pushConfig for this cacheKey, and
the caller
+// should leave its status untouched rather than treat the absence of an
execution error
+// as success. A non-nil result, empty or not, means pushConfig actually ran.
func (d *apisixProvider) syncConfigNow(
ctx context.Context,
name string,
build func() (adcclient.SyncInput, error),
-) (types.ADCExecutionErrors, error) {
+) (result *types.ADCExecutionErrors, err error) {
unlock := d.syncLocks.Lock(name)
defer unlock()
input, err := build()
if err != nil {
- return types.ADCExecutionErrors{}, err
+ return nil, err
}
execErrs := d.pushConfig(ctx, input)
if len(execErrs.Errors) > 0 {
- return execErrs, execErrs
+ return &execErrs, execErrs
}
- return execErrs, nil
+ return &execErrs, nil
}
// pushConfig sends input to its data plane and shapes whatever failed into
the form
@@ -409,13 +417,13 @@ func toADCExecutionError(name string, err error)
types.ADCExecutionError {
}
// syncEvictedConfigsNow pushes an empty resource set for each of the given
configs
-// immediately, instead of waiting for the next scheduled sync round --
through the same
+// immediately, instead of waiting for the next scheduled sync round, through
the same
// per-cacheKey lock the periodic sync uses, so it can never race a periodic
round for the
-// same GatewayProxy. Used only when the deleted resource is a Gateway or
IngressClass --
-// resourceTypes is empty for those, so the preceding removeResourceState call
already
+// same GatewayProxy. Used only when the deleted resource is a Gateway or
IngressClass,
+// where resourceTypes is empty, so the preceding removeResourceState call
already
// reset each config's whole cached snapshot via Store.Delete, and that reset
should reach
-// the data plane promptly. Failures are logged, not surfaced as a status
update -- this
-// mirrors the deferred path, which only reports through the next scheduled
sync round.
+// the data plane promptly. Failures are logged, not surfaced as a status
update, matching
+// the deferred path, which only reports through the next scheduled sync round.
func (d *apisixProvider) syncEvictedConfigsNow(
ctx context.Context,
configs map[types.NamespacedNameKind]adctypes.Config,
@@ -493,34 +501,40 @@ func (d *apisixProvider) Start(ctx context.Context) error
{
}
}
-// sync pushes every GatewayProxy AIC currently knows about, config by config
-- each
-// one's current resource snapshot is only read once syncConfigNow actually
holds that
-// cacheKey's lock, so a slow round can never push a snapshot that was already
stale by the
-// time its turn came up. All of this round's results are still collected into
one
-// statusesMap and handed to handleADCExecutionErrors together, exactly as a
single batched
-// sync would: that logic diffs against last round's full picture, not
per-config.
+// sync pushes every GatewayProxy AIC currently knows about, config by config,
each one's
+// current resource snapshot is only read once syncConfigNow actually holds
that
+// cacheKey's lock, so a slow round can never push a snapshot that was already
stale by
+// the time its turn came up. results collects one entry per config this round
actually
+// reached pushConfig for, success (a zero-value types.ADCExecutionErrors) or
failure. A
+// config whose build itself failed (a local error, before anything reached
the data
+// plane) is left out of results entirely and its status goes untouched this
round,
+// logged here rather than silently treated as either outcome; see
+// updateStatusFromSyncResults for what results feeds into.
func (d *apisixProvider) sync(ctx context.Context) error {
configs := d.configManager.List()
- statusesMap := map[string]types.ADCExecutionErrors{}
+ results := map[string]types.ADCExecutionErrors{}
var errs []error
for _, config := range configs {
- execErrs, err := d.syncConfigNow(ctx, config.Name, func()
(adcclient.SyncInput, error) {
+ result, err := d.syncConfigNow(ctx, config.Name, func()
(adcclient.SyncInput, error) {
resources, err := d.store.GetResources(config.Name)
if err != nil {
return adcclient.SyncInput{},
fmt.Errorf("failed to get resources from store: %w", err)
}
return adcclient.SyncInput{Name: config.Name, Config:
config, Resources: resources}, nil
})
- if err != nil {
+ if result == nil {
+ d.log.Error(err, "failed to build sync input, leaving
this GatewayProxy's status untouched this round", "config", config.Name)
errs = append(errs, fmt.Errorf("config %s: %w",
config.Name, err))
+ continue
}
- if len(execErrs.Errors) > 0 {
- statusesMap[config.Name] = execErrs
+ results[config.Name] = *result
+ if err != nil {
+ errs = append(errs, fmt.Errorf("config %s: %w",
config.Name, err))
}
}
- d.handleADCExecutionErrors(statusesMap)
+ d.updateStatusFromSyncResults(ctx, results)
return errors.Join(errs...)
}
@@ -531,12 +545,6 @@ func (d *apisixProvider) syncNotify() {
}
}
-func (d *apisixProvider) handleADCExecutionErrors(statusesMap
map[string]types.ADCExecutionErrors) {
- statusUpdateMap := d.resolveADCExecutionErrors(statusesMap)
- d.handleStatusUpdate(statusUpdateMap)
- d.log.V(1).Info("handled ADC execution errors", "status_record",
statusesMap, "status_update", statusUpdateMap)
-}
-
func (d *apisixProvider) NeedLeaderElection() bool {
return true
}
diff --git a/internal/provider/apisix/status.go
b/internal/provider/apisix/status.go
index e2f82bd0..0aeb58ce 100644
--- a/internal/provider/apisix/status.go
+++ b/internal/provider/apisix/status.go
@@ -18,13 +18,17 @@
package apisix
import (
+ "context"
"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"
+ adctypes "github.com/apache/apisix-ingress-controller/api/adc"
+ apiv1alpha1 "github.com/apache/apisix-ingress-controller/api/v1alpha1"
apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
"github.com/apache/apisix-ingress-controller/internal/controller/label"
"github.com/apache/apisix-ingress-controller/internal/controller/status"
@@ -32,42 +36,217 @@ import (
"github.com/apache/apisix-ingress-controller/internal/types"
)
-// handleStatusUpdate updates resource conditions based on the latest sync
results.
+// GatewayProxyConditionDataPlaneAvailable reports whether every APISIX
instance a
+// GatewayProxy addresses took the last sync. It lands on the GatewayProxy
rather than
+// a Gateway, an IngressClass, or a CRD, since GatewayProxy is what every path
shares.
+const (
+ GatewayProxyConditionDataPlaneAvailable = "DataPlaneAvailable"
+
+ GatewayProxyReasonDataPlaneAvailable = "DataPlaneAvailable"
+ GatewayProxyReasonDataPlaneInstanceUnavailable =
"DataPlaneInstanceUnavailable"
+)
+
+// updateStatusFromSyncResults updates every resource and GatewayProxy status
this
+// round's sync results call for. results holds one entry per config sync()
actually
+// reached pushConfig for this round, success (a zero-value
types.ADCExecutionErrors) or
+// failure; a config sync() could not even build a SyncInput for is absent
here entirely
+// and its status is left untouched this round, see sync().
//
-// It maintains a history of failed resources in d.statusUpdateMap.
+// GatewayProxy's DataPlaneAvailable condition is recomputed and written fresh
every
+// round directly from this round's result, never compared against any
remembered
+// history: a config with no error this round is written True, one with any
error is
+// written False. That is what makes a GatewayProxy that has always been
healthy actually
+// get a True the first time, and what keeps a restart from leaving a stale
False stuck
+// forever: the write only ever depends on this round's actual outcome.
//
-// For resources in the current failure map (statusUpdateMap), it marks them
as failed.
-// For resources that exist only in the previous failure history (i.e. not in
this sync's failures),
-// it marks them as accepted (success).
-func (d *apisixProvider) handleStatusUpdate(statusUpdateMap
map[types.NamespacedNameKind][]string) {
- // Mark all resources in the current failure set as failed.
- for nnk, msgs := range statusUpdateMap {
- d.updateStatus(nnk, cutils.NewConditionTypeAccepted(
- apiv2.ConditionReasonSyncFailed,
- false,
- 0,
- strings.Join(msgs, "; "),
- ))
+// Resource status can't afford the same full recompute: a config's resource
set can be
+// large, and rewriting every one of them every round even when nothing
changed would be
+// wasteful. So resources keep a small persisted delta in d.resourceFailures
instead:
+// newly (or still) failing resources are written SyncFailed, and any resource
that was
+// failing last round but isn't failing this one gets its error explicitly
cleared with
+// an Accepted write.
+func (d *apisixProvider) updateStatusFromSyncResults(ctx context.Context,
results map[string]types.ADCExecutionErrors) {
+ resourceFailures := map[types.NamespacedNameKind][]string{}
+
+ for configName, execErrs := range results {
+ var gatewayProxy types.NamespacedNameKind
+ if err := gatewayProxy.FromString(configName); err != nil {
+ d.log.Error(err, "failed to parse config name as a
GatewayProxy key", "configName", configName)
+ continue
+ }
+
+ gatewayProxyMsgs, failedEndpoints :=
d.classifySyncResult(configName, execErrs, resourceFailures)
+ if len(gatewayProxyMsgs) > 0 {
+ d.updateStatus(gatewayProxy,
failureCondition(gatewayProxy, strings.Join(gatewayProxyMsgs, "; ")))
+ d.recordFailedEndpointEvents(ctx, gatewayProxy,
failedEndpoints)
+ } else {
+ d.updateStatus(gatewayProxy,
successCondition(gatewayProxy))
+ }
+ }
+
+ d.applyResourceFailures(resourceFailures)
+ d.log.V(1).Info("updated status from sync results", "results", results,
"resource_failures", resourceFailures)
+}
+
+// classifySyncResult splits one config's this-round result into what belongs
on the
+// GatewayProxy (returned) and what belongs on specific Kubernetes resources
(added into
+// resourceFailures). The two are independent, not mutually exclusive: a
single addrErr
+// can carry both a resource-attributed FailedStatuses entry and a failed
+// EndpointStatuses entry at once (apisix-standalone's own re-validate path
attaches
+// EndpointStatuses to every addrErr regardless of what FailedStatuses also
names), so
+// checking one must never suppress reporting the other. A FailedStatuses
entry that
+// resolves to a resource via its Event goes there; everything else, no
FailedStatuses at
+// all, or a FailedStatuses entry with no Event to resolve (apisix-standalone
when the
+// rejection can't be pinned on a specific resource), is a GatewayProxy-level
signal
+// instead, but only once nothing else already explained this addrErr:
EndpointStatuses'
+// own message first, the raw error as a last resort.
+func (d *apisixProvider) classifySyncResult(
+ configName string,
+ execErrs types.ADCExecutionErrors,
+ resourceFailures map[types.NamespacedNameKind][]string,
+) (gatewayProxyMsgs []string, failedEndpoints []adctypes.EndpointStatus) {
+ for _, execErr := range execErrs.Errors {
+ for _, addrErr := range execErr.FailedErrors {
+ endpointMsg :=
unavailableEndpointsMessage(addrErr.EndpointStatuses)
+ if endpointMsg != "" {
+ gatewayProxyMsgs = append(gatewayProxyMsgs,
endpointMsg)
+ failedEndpoints = append(failedEndpoints,
addrErr.EndpointStatuses...)
+ }
+
+ if len(addrErr.FailedStatuses) == 0 {
+ if endpointMsg == "" {
+ gatewayProxyMsgs =
append(gatewayProxyMsgs, addrErr.Error())
+ }
+ continue
+ }
+
+ anyUnattributed := false
+ for _, syncStatus := range addrErr.FailedStatuses {
+ if syncStatus.Event.ResourceType == "" {
+ anyUnattributed = true
+ continue
+ }
+ labels, err :=
d.store.GetResourceLabel(configName, syncStatus.Event.ResourceType,
syncStatus.Event.ResourceID)
+ if err != nil {
+ d.log.Error(err, "failed to get
resource label",
+ "configName", configName,
"resourceType", syncStatus.Event.ResourceType, "id",
syncStatus.Event.ResourceID)
+ continue
+ }
+ resourceKey := types.NamespacedNameKind{
+ Name: labels[label.LabelName],
+ Namespace: labels[label.LabelNamespace],
+ Kind: labels[label.LabelKind],
+ }
+ msg := fmt.Sprintf("ServerAddr: %s, Error: %s",
addrErr.ServerAddr, syncStatus.Reason)
+ resourceFailures[resourceKey] =
append(resourceFailures[resourceKey], msg)
+ }
+ if anyUnattributed && endpointMsg == "" {
+ gatewayProxyMsgs = append(gatewayProxyMsgs,
addrErr.Error())
+ }
+ }
+ }
+ return gatewayProxyMsgs, failedEndpoints
+}
+
+// applyResourceFailures writes this round's newly (or still) failing
resources, and
+// clears the recorded error from any resource that was failing last round but
isn't in
+// newFailures now. See updateStatusFromSyncResults for why resources use this
delta
+// instead of GatewayProxy's full recompute.
+func (d *apisixProvider) applyResourceFailures(newFailures
map[types.NamespacedNameKind][]string) {
+ for resourceKey, msgs := range newFailures {
+ d.updateStatus(resourceKey, failureCondition(resourceKey,
strings.Join(msgs, "; ")))
+ }
+ for resourceKey := range d.resourceFailures {
+ if _, stillFailing := newFailures[resourceKey]; !stillFailing {
+ d.updateStatus(resourceKey,
successCondition(resourceKey))
+ }
+ }
+ d.resourceFailures = newFailures
+}
+
+// failureCondition and successCondition pick which condition a
NamespacedNameKind
+// gets: GatewayProxyConditionDataPlaneAvailable for a GatewayProxy, the
existing
+// Accepted/SyncFailed condition for everything else.
+func failureCondition(nnk types.NamespacedNameKind, msg string)
metav1.Condition {
+ if nnk.Kind == types.KindGatewayProxy {
+ return newGatewayProxyDataPlaneAvailableCondition(false,
GatewayProxyReasonDataPlaneInstanceUnavailable, msg)
+ }
+ return cutils.NewConditionTypeAccepted(apiv2.ConditionReasonSyncFailed,
false, 0, msg)
+}
+
+func successCondition(nnk types.NamespacedNameKind) metav1.Condition {
+ if nnk.Kind == types.KindGatewayProxy {
+ return newGatewayProxyDataPlaneAvailableCondition(true,
GatewayProxyReasonDataPlaneAvailable, "")
+ }
+ return cutils.NewConditionTypeAccepted(apiv2.ConditionReasonAccepted,
true, 0, "")
+}
+
+func newGatewayProxyDataPlaneAvailableCondition(available bool, reason, msg
string) metav1.Condition {
+ conditionStatus := metav1.ConditionFalse
+ if available {
+ conditionStatus = metav1.ConditionTrue
+ }
+ return metav1.Condition{
+ Type: GatewayProxyConditionDataPlaneAvailable,
+ Status: conditionStatus,
+ LastTransitionTime: metav1.Now(),
+ Reason: reason,
+ Message: cutils.TruncateConditionMessage(msg),
+ }
+}
+
+// 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. The GatewayProxy is fetched fresh
from the
+// API server first so the Event's involvedObject carries a real UID: a
hand-built stub
+// with only Name/Namespace leaves that UID empty, and kubectl describe
resolves events
+// by matching it, so an event against such a stub never shows up there.
+func (d *apisixProvider) recordFailedEndpointEvents(ctx context.Context, nnk
types.NamespacedNameKind, endpoints []adctypes.EndpointStatus) {
+ if d.EventRecorder == nil {
+ return
+ }
+ hasFailure := false
+ for _, ep := range endpoints {
+ if !ep.Success {
+ hasFailure = true
+ break
+ }
+ }
+ if !hasFailure {
+ return
}
- // Mark resources that exist only in the previous failure history as
successful.
- for nnk := range d.statusUpdateMap {
- if _, ok := statusUpdateMap[nnk]; !ok {
- d.updateStatus(nnk, cutils.NewConditionTypeAccepted(
- apiv2.ConditionReasonAccepted,
- true,
- 0,
- "",
- ))
+ gatewayProxy := &apiv1alpha1.GatewayProxy{}
+ if err := d.K8sClient.Get(ctx, nnk.NamespacedName(), gatewayProxy); err
!= nil {
+ d.log.Error(err, "failed to get GatewayProxy to record failed
endpoint events", "name", nnk.Name, "namespace", nnk.Namespace)
+ return
+ }
+
+ for _, ep := range endpoints {
+ if ep.Success {
+ continue
}
+ d.EventRecorder.Event(gatewayProxy, corev1.EventTypeWarning,
GatewayProxyReasonDataPlaneInstanceUnavailable,
+ fmt.Sprintf("%s: %s", ep.Server, ep.Reason))
}
- // Update the failure history with the current failure set.
- d.statusUpdateMap = statusUpdateMap
}
//nolint:gocyclo
func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition
metav1.Condition) {
switch nnk.Kind {
+ case types.KindGatewayProxy:
+ // Unlike the route kinds below, the condition lands on the
GatewayProxy's own
+ // top-level Status.Conditions, not on a per-parent entry.
+ d.updater.Update(status.Update{
+ NamespacedName: nnk.NamespacedName(),
+ Resource: &apiv1alpha1.GatewayProxy{},
+ Mutator: status.MutatorFunc(func(obj client.Object)
client.Object {
+ cp := obj.(*apiv1alpha1.GatewayProxy).DeepCopy()
+ condition.ObservedGeneration =
cp.GetGeneration()
+ cp.Status.Conditions =
cutils.MergeCondition(cp.Status.Conditions, condition)
+ return cp
+ }),
+ })
case types.KindApisixRoute:
d.updater.Update(status.Update{
NamespacedName: nnk.NamespacedName(),
@@ -255,96 +434,19 @@ func (d *apisixProvider) updateStatus(nnk
types.NamespacedNameKind, condition me
}
}
-func (d *apisixProvider) resolveADCExecutionErrors(
- statusesMap map[string]types.ADCExecutionErrors,
-) map[types.NamespacedNameKind][]string {
- statusUpdateMap := map[types.NamespacedNameKind][]string{}
- for configName, execErrors := range statusesMap {
- for _, execErr := range execErrors.Errors {
- for _, failedStatus := range execErr.FailedErrors {
- if len(failedStatus.FailedStatuses) == 0 {
- d.handleEmptyFailedStatuses(configName,
failedStatus, statusUpdateMap)
- } else {
-
d.handleDetailedFailedStatuses(configName, failedStatus, statusUpdateMap)
- }
- }
- }
- }
-
- return statusUpdateMap
-}
-
-func (d *apisixProvider) handleEmptyFailedStatuses(
- configName string,
- failedStatus types.ADCExecutionServerAddrError,
- statusUpdateMap map[types.NamespacedNameKind][]string,
-) {
- resource, err := d.store.GetResources(configName)
- if err != nil {
- d.log.Error(err, "failed to get resources from store",
"configName", configName)
- return
- }
-
- for _, obj := range resource.Services {
- d.addResourceToStatusUpdateMap(obj.GetLabels(),
failedStatus.Error(), statusUpdateMap)
- }
-
- for _, obj := range resource.Consumers {
- d.addResourceToStatusUpdateMap(obj.GetLabels(),
failedStatus.Error(), statusUpdateMap)
- }
-
- for _, obj := range resource.SSLs {
- d.addResourceToStatusUpdateMap(obj.GetLabels(),
failedStatus.Error(), statusUpdateMap)
- }
-
- globalRules, err := d.store.ListGlobalRules(configName)
- if err != nil {
- d.log.Error(err, "failed to list global rules", "configName",
configName)
- return
- }
- for _, rule := range globalRules {
- d.addResourceToStatusUpdateMap(rule.GetLabels(),
failedStatus.Error(), statusUpdateMap)
- }
-}
-
-func (d *apisixProvider) handleDetailedFailedStatuses(
- configName string,
- failedStatus types.ADCExecutionServerAddrError,
- statusUpdateMap map[types.NamespacedNameKind][]string,
-) {
- for _, status := range failedStatus.FailedStatuses {
- // in the APISIX standalone mode, the related values in the
sync failure event are empty.
- if status.Event.ResourceType == "" {
- d.handleEmptyFailedStatuses(configName, failedStatus,
statusUpdateMap)
- return
- }
- id := status.Event.ResourceID
- labels, err := d.store.GetResourceLabel(configName,
status.Event.ResourceType, id)
- if err != nil {
- d.log.Error(err, "failed to get resource label",
- "configName", configName,
- "resourceType", status.Event.ResourceType,
- "id", id,
- )
+// unavailableEndpointsMessage summarizes every EndpointStatus entry that
didn't
+// succeed, in the order given. Empty means none did (or there were none to
check).
+func unavailableEndpointsMessage(endpoints []adctypes.EndpointStatus) string {
+ failed := make([]string, 0, len(endpoints))
+ for _, ep := range endpoints {
+ if ep.Success {
continue
}
- d.addResourceToStatusUpdateMap(
- labels,
- fmt.Sprintf("ServerAddr: %s, Error: %s",
failedStatus.ServerAddr, status.Reason),
- statusUpdateMap,
- )
+ failed = append(failed, fmt.Sprintf("%s: %s", ep.Server,
ep.Reason))
}
-}
-
-func (d *apisixProvider) addResourceToStatusUpdateMap(
- labels map[string]string,
- msg string,
- statusUpdateMap map[types.NamespacedNameKind][]string,
-) {
- statusKey := types.NamespacedNameKind{
- Name: labels[label.LabelName],
- Namespace: labels[label.LabelNamespace],
- Kind: labels[label.LabelKind],
+ if len(failed) == 0 {
+ return ""
}
- statusUpdateMap[statusKey] = append(statusUpdateMap[statusKey], msg)
+ return fmt.Sprintf("%d/%d gateway instance(s) failed to apply the last
sync: %s",
+ len(failed), len(endpoints), strings.Join(failed, "; "))
}
diff --git a/internal/provider/apisix/status_test.go
b/internal/provider/apisix/status_test.go
new file mode 100644
index 00000000..81fa567d
--- /dev/null
+++ b/internal/provider/apisix/status_test.go
@@ -0,0 +1,437 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package apisix
+
+import (
+ "context"
+ "strings"
+ "testing"
+
+ "github.com/go-logr/logr"
+ "github.com/stretchr/testify/require"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/runtime"
+ clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+ "k8s.io/client-go/tools/record"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+ "sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+ adctypes "github.com/apache/apisix-ingress-controller/api/adc"
+ apiv1alpha1 "github.com/apache/apisix-ingress-controller/api/v1alpha1"
+ apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
+ "github.com/apache/apisix-ingress-controller/internal/adc/cache"
+ "github.com/apache/apisix-ingress-controller/internal/controller/label"
+ "github.com/apache/apisix-ingress-controller/internal/controller/status"
+ "github.com/apache/apisix-ingress-controller/internal/types"
+)
+
+// fakeK8sClient builds a controller-runtime fake client seeded with objects,
for the
+// GatewayProxy lookup recordFailedEndpointEvents does before firing an Event.
+func fakeK8sClient(t *testing.T, objects ...runtime.Object) client.Client {
+ t.Helper()
+ scheme := runtime.NewScheme()
+ require.NoError(t, clientgoscheme.AddToScheme(scheme))
+ require.NoError(t, apiv1alpha1.AddToScheme(scheme))
+ return
fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(objects...).Build()
+}
+
+func TestUnavailableEndpointsMessageEmptyWhenEverythingSucceeded(t *testing.T)
{
+ msg := unavailableEndpointsMessage([]adctypes.EndpointStatus{
+ {Server: "http://apisix-1:9180";, Success: true},
+ {Server: "http://apisix-2:9180";, Success: true},
+ })
+ if msg != "" {
+ t.Fatalf("expected no message, got %q", msg)
+ }
+}
+
+func TestUnavailableEndpointsMessageEmptyWhenThereAreNoEndpoints(t *testing.T)
{
+ if msg := unavailableEndpointsMessage(nil); msg != "" {
+ t.Fatalf("expected no message, got %q", msg)
+ }
+}
+
+func TestUnavailableEndpointsMessageSummarizesOnlyTheFailedOnes(t *testing.T) {
+ msg := unavailableEndpointsMessage([]adctypes.EndpointStatus{
+ {Server: "http://apisix-1:9180";, Success: true},
+ {Server: "http://apisix-2:9180";, Success: false, Reason:
"connection refused"},
+ })
+ want := "1/2 gateway instance(s) failed to apply the last sync:
http://apisix-2:9180: connection refused"
+ if msg != want {
+ t.Fatalf("got %q, want %q", msg, want)
+ }
+}
+
+func TestFailureConditionMarksGatewayProxyDataPlaneAvailableFalse(t
*testing.T) {
+ nnk := types.NamespacedNameKind{Kind: types.KindGatewayProxy,
Namespace: "ns", Name: "gp"}
+ c := failureCondition(nnk, "boom")
+
+ if c.Type != GatewayProxyConditionDataPlaneAvailable {
+ t.Errorf("Type = %q, want %q", c.Type,
GatewayProxyConditionDataPlaneAvailable)
+ }
+ if c.Status != metav1.ConditionFalse {
+ t.Errorf("Status = %v, want False", c.Status)
+ }
+ if c.Reason != GatewayProxyReasonDataPlaneInstanceUnavailable {
+ t.Errorf("Reason = %q, want %q", c.Reason,
GatewayProxyReasonDataPlaneInstanceUnavailable)
+ }
+ if c.Message != "boom" {
+ t.Errorf("Message = %q, want %q", c.Message, "boom")
+ }
+}
+
+func TestSuccessConditionMarksGatewayProxyDataPlaneAvailableTrue(t *testing.T)
{
+ nnk := types.NamespacedNameKind{Kind: types.KindGatewayProxy,
Namespace: "ns", Name: "gp"}
+ c := successCondition(nnk)
+
+ if c.Type != GatewayProxyConditionDataPlaneAvailable {
+ t.Errorf("Type = %q, want %q", c.Type,
GatewayProxyConditionDataPlaneAvailable)
+ }
+ if c.Status != metav1.ConditionTrue {
+ t.Errorf("Status = %v, want True", c.Status)
+ }
+ if c.Reason != GatewayProxyReasonDataPlaneAvailable {
+ t.Errorf("Reason = %q, want %q", c.Reason,
GatewayProxyReasonDataPlaneAvailable)
+ }
+}
+
+func TestFailureAndSuccessConditionKeepUsingAcceptedForNonGatewayProxyKinds(t
*testing.T) {
+ nnk := types.NamespacedNameKind{Kind: types.KindApisixRoute, Namespace:
"ns", Name: "route"}
+
+ failed := failureCondition(nnk, "boom")
+ if failed.Type != string(apiv2.ConditionTypeAccepted) {
+ t.Errorf("failure Type = %q, want %q", failed.Type,
apiv2.ConditionTypeAccepted)
+ }
+ if failed.Reason != string(apiv2.ConditionReasonSyncFailed) {
+ t.Errorf("failure Reason = %q, want %q", failed.Reason,
apiv2.ConditionReasonSyncFailed)
+ }
+
+ ok := successCondition(nnk)
+ if ok.Type != string(apiv2.ConditionTypeAccepted) {
+ t.Errorf("success Type = %q, want %q", ok.Type,
apiv2.ConditionTypeAccepted)
+ }
+ if ok.Reason != string(apiv2.ConditionReasonAccepted) {
+ t.Errorf("success Reason = %q, want %q", ok.Reason,
apiv2.ConditionReasonAccepted)
+ }
+}
+
+func TestRecordFailedEndpointEventsFiresOneWarningPerFailedEndpoint(t
*testing.T) {
+ recorder := record.NewFakeRecorder(2)
+ d := &apisixProvider{log: logr.Discard()}
+ d.EventRecorder = recorder
+ d.K8sClient = fakeK8sClient(t, &apiv1alpha1.GatewayProxy{
+ ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "gp", UID:
"gp-uid"},
+ })
+ nnk := types.NamespacedNameKind{Kind: types.KindGatewayProxy,
Namespace: "ns", Name: "gp"}
+
+ d.recordFailedEndpointEvents(context.Background(), 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. No K8sClient
either: a nil
+ // dereference here would mean this didn't actually return before
reaching it.
+ d.recordFailedEndpointEvents(context.Background(), nnk,
[]adctypes.EndpointStatus{{Server: "http://apisix-1:9180";, Success: false}})
+}
+
+func TestRecordFailedEndpointEventsNoopsWhenEveryEndpointSucceeded(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"}
+
+ // No K8sClient configured either: with nothing failed, the
GatewayProxy lookup this
+ // needs to attribute a failure must never even be attempted.
+ d.recordFailedEndpointEvents(context.Background(), nnk,
[]adctypes.EndpointStatus{{Server: "http://apisix-1:9180";, Success: true}})
+
+ select {
+ case e := <-recorder.Events:
+ t.Errorf("unexpected event %q, nothing failed", e)
+ default:
+ }
+}
+
+func TestRecordFailedEndpointEventsSkipsWhenTheGatewayProxyCannotBeFetched(t
*testing.T) {
+ recorder := record.NewFakeRecorder(1)
+ d := &apisixProvider{log: logr.Discard()}
+ d.EventRecorder = recorder
+ d.K8sClient = fakeK8sClient(t) // no GatewayProxy seeded, so Get
returns NotFound
+ nnk := types.NamespacedNameKind{Kind: types.KindGatewayProxy,
Namespace: "ns", Name: "gp"}
+
+ // Must not panic, and must not fire an event against a zero-value
stand-in.
+ d.recordFailedEndpointEvents(context.Background(), nnk,
[]adctypes.EndpointStatus{{Server: "http://apisix-1:9180";, Success: false}})
+
+ select {
+ case e := <-recorder.Events:
+ t.Errorf("unexpected event %q, the GatewayProxy could not be
fetched", e)
+ default:
+ }
+}
+
+// fakeUpdater records every status.Update handed to it, and applies each
Mutator against
+// a caller-supplied base object so a test can inspect the condition it
actually wrote.
+type fakeUpdater struct {
+ updates []status.Update
+}
+
+func (f *fakeUpdater) Update(u status.Update) {
+ f.updates = append(f.updates, u)
+}
+
+func TestClassifySyncResultHardErrorGoesToGatewayProxy(t *testing.T) {
+ d := &apisixProvider{log: logr.Discard()}
+ execErrs := types.ADCExecutionErrors{Errors: []types.ADCExecutionError{{
+ Name: "GatewayProxy/ns/gp",
+ FailedErrors: []types.ADCExecutionServerAddrError{{
+ ServerAddr: "http://apisix:9180";,
+ Err: "HTTP 500: boom",
+ }},
+ }}}
+
+ resourceFailures := map[types.NamespacedNameKind][]string{}
+ gatewayProxyMsgs, failedEndpoints :=
d.classifySyncResult("GatewayProxy/ns/gp", execErrs, resourceFailures)
+
+ if len(resourceFailures) != 0 {
+ t.Errorf("expected no resource attributed, got %v",
resourceFailures)
+ }
+ if len(failedEndpoints) != 0 {
+ t.Errorf("expected no endpoints, got %v", failedEndpoints)
+ }
+ if len(gatewayProxyMsgs) != 1 || !strings.Contains(gatewayProxyMsgs[0],
"HTTP 500: boom") {
+ t.Errorf("gatewayProxyMsgs = %v, want the raw error",
gatewayProxyMsgs)
+ }
+}
+
+func TestClassifySyncResultEndpointFailuresGoToGatewayProxy(t *testing.T) {
+ d := &apisixProvider{log: logr.Discard()}
+ endpoints := []adctypes.EndpointStatus{
+ {Server: "http://apisix-1:9180";, Success: true},
+ {Server: "http://apisix-2:9180";, Success: false, Reason:
"connection refused"},
+ }
+ execErrs := types.ADCExecutionErrors{Errors: []types.ADCExecutionError{{
+ Name: "GatewayProxy/ns/gp",
+ FailedErrors: []types.ADCExecutionServerAddrError{{
+ EndpointStatuses: endpoints,
+ }},
+ }}}
+
+ resourceFailures := map[types.NamespacedNameKind][]string{}
+ gatewayProxyMsgs, failedEndpoints :=
d.classifySyncResult("GatewayProxy/ns/gp", execErrs, resourceFailures)
+
+ if len(resourceFailures) != 0 {
+ t.Errorf("expected no resource attributed, got %v",
resourceFailures)
+ }
+ if len(gatewayProxyMsgs) != 1 || !strings.Contains(gatewayProxyMsgs[0],
"http://apisix-2:9180: connection refused") {
+ t.Errorf("gatewayProxyMsgs = %v, want the endpoint summary",
gatewayProxyMsgs)
+ }
+ if len(failedEndpoints) != 2 {
+ t.Errorf("failedEndpoints = %v, want every EndpointStatus entry
passed through for event firing", failedEndpoints)
+ }
+}
+
+func TestClassifySyncResultAttributesFailedStatusesToTheirResource(t
*testing.T) {
+ d := &apisixProvider{log: logr.Discard(), store:
cache.NewStore(logr.Discard())}
+ const configName = "GatewayProxy/ns/gp"
+ if err := d.store.Insert(configName, []string{adctypes.TypeService},
&adctypes.Resources{
+ Services: []*adctypes.Service{{
+ Metadata: adctypes.Metadata{
+ ID: "svc1",
+ Labels: map[string]string{
+ label.LabelKind: "ApisixRoute",
+ label.LabelName: "route1",
+ label.LabelNamespace: "ns1",
+ },
+ },
+ }},
+ }, nil); err != nil {
+ t.Fatalf("seeding the store: %v", err)
+ }
+
+ execErrs := types.ADCExecutionErrors{Errors: []types.ADCExecutionError{{
+ Name: configName,
+ FailedErrors: []types.ADCExecutionServerAddrError{{
+ ServerAddr: "http://apisix:9180";,
+ FailedStatuses: []adctypes.SyncStatus{{
+ Reason: "unknown plugin foo",
+ Event: adctypes.StatusEvent{ResourceType:
adctypes.TypeService, ResourceID: "svc1"},
+ }},
+ }},
+ }}}
+
+ resourceFailures := map[types.NamespacedNameKind][]string{}
+ gatewayProxyMsgs, _ := d.classifySyncResult(configName, execErrs,
resourceFailures)
+
+ if len(gatewayProxyMsgs) != 0 {
+ t.Errorf("expected nothing attributed to the GatewayProxy, got
%v", gatewayProxyMsgs)
+ }
+ want := types.NamespacedNameKind{Kind: "ApisixRoute", Namespace: "ns1",
Name: "route1"}
+ if got := resourceFailures[want]; len(got) != 1 ||
!strings.Contains(got[0], "unknown plugin foo") {
+ t.Errorf("resourceFailures[%v] = %v, want the failure reason",
want, got)
+ }
+}
+
+func
TestClassifySyncResultReportsEndpointStatusesEvenOnAFullyAttributedAddrErr(t
*testing.T) {
+ // apisix-standalone's own re-validate path attaches EndpointStatuses
to every addrErr
+ // regardless of whether FailedStatuses also named specific resources:
an all-rejected
+ // write leaves every endpoint success:false, and one of those
endpoints may be
+ // failing for a reason that has nothing to do with the resource
FailedStatuses names
+ // (e.g. genuinely unreachable, not just rejecting this content). The
two signals are
+ // independent: attributing the resource failure must never suppress
reporting the
+ // endpoint failure too.
+ d := &apisixProvider{log: logr.Discard(), store:
cache.NewStore(logr.Discard())}
+ const configName = "GatewayProxy/ns/gp"
+ if err := d.store.Insert(configName, []string{adctypes.TypeService},
&adctypes.Resources{
+ Services: []*adctypes.Service{{
+ Metadata: adctypes.Metadata{
+ ID: "svc1",
+ Labels: map[string]string{
+ label.LabelKind: "ApisixRoute",
+ label.LabelName: "route1",
+ label.LabelNamespace: "ns1",
+ },
+ },
+ }},
+ }, nil); err != nil {
+ t.Fatalf("seeding the store: %v", err)
+ }
+
+ execErrs := types.ADCExecutionErrors{Errors: []types.ADCExecutionError{{
+ Name: configName,
+ FailedErrors: []types.ADCExecutionServerAddrError{{
+ ServerAddr: "http://apisix:9180";,
+ FailedStatuses: []adctypes.SyncStatus{{
+ Reason: "unknown plugin foo",
+ Event: adctypes.StatusEvent{ResourceType:
adctypes.TypeService, ResourceID: "svc1"},
+ }},
+ EndpointStatuses: []adctypes.EndpointStatus{
+ {Server: "http://apisix-1:9180";, Success:
false, Reason: "content rejected"},
+ {Server: "http://apisix-2:9180";, Success:
false, Reason: "connection refused"},
+ },
+ }},
+ }}}
+
+ resourceFailures := map[types.NamespacedNameKind][]string{}
+ gatewayProxyMsgs, failedEndpoints := d.classifySyncResult(configName,
execErrs, resourceFailures)
+
+ if len(gatewayProxyMsgs) != 1 || !strings.Contains(gatewayProxyMsgs[0],
"http://apisix-2:9180: connection refused") {
+ t.Errorf("gatewayProxyMsgs = %v, want the endpoint summary",
gatewayProxyMsgs)
+ }
+ if len(failedEndpoints) != 2 {
+ t.Errorf("failedEndpoints = %v, want every EndpointStatus entry
passed through for event firing", failedEndpoints)
+ }
+ want := types.NamespacedNameKind{Kind: "ApisixRoute", Namespace: "ns1",
Name: "route1"}
+ if got := resourceFailures[want]; len(got) != 1 ||
!strings.Contains(got[0], "unknown plugin foo") {
+ t.Errorf("resourceFailures[%v] = %v, want the failure reason",
want, got)
+ }
+}
+
+func
TestClassifySyncResultFallsBackToGatewayProxyWhenAFailedStatusHasNoResourceAttribution(t
*testing.T) {
+ // apisix-standalone: FailedStatuses can be non-empty yet carry no
Event to resolve a
+ // resource from at all, the whole addrErr is then a GatewayProxy-level
signal.
+ d := &apisixProvider{log: logr.Discard()}
+ execErrs := types.ADCExecutionErrors{Errors: []types.ADCExecutionError{{
+ Name: "GatewayProxy/ns/gp",
+ FailedErrors: []types.ADCExecutionServerAddrError{{
+ Err: "all_failed",
+ FailedStatuses: []adctypes.SyncStatus{{Reason: "schema
error"}},
+ }},
+ }}}
+
+ resourceFailures := map[types.NamespacedNameKind][]string{}
+ gatewayProxyMsgs, _ := d.classifySyncResult("GatewayProxy/ns/gp",
execErrs, resourceFailures)
+
+ if len(resourceFailures) != 0 {
+ t.Errorf("expected no resource attributed, got %v",
resourceFailures)
+ }
+ if len(gatewayProxyMsgs) != 1 {
+ t.Errorf("gatewayProxyMsgs = %v, want exactly one fallback
message", gatewayProxyMsgs)
+ }
+}
+
+func TestApplyResourceFailuresWritesNewFailuresAndClearsResolvedOnes(t
*testing.T) {
+ updater := &fakeUpdater{}
+ d := &apisixProvider{
+ log: logr.Discard(),
+ updater: updater,
+ resourceFailures: map[types.NamespacedNameKind][]string{
+ {Kind: types.KindApisixRoute, Namespace: "ns", Name:
"resolved"}: {"used to fail"},
+ {Kind: types.KindApisixRoute, Namespace: "ns", Name:
"still-bad"}: {"still failing"},
+ },
+ }
+
+ newFailures := map[types.NamespacedNameKind][]string{
+ {Kind: types.KindApisixRoute, Namespace: "ns", Name:
"still-bad"}: {"still failing"},
+ {Kind: types.KindApisixRoute, Namespace: "ns", Name:
"newly-bad"}: {"new failure"},
+ }
+
+ d.applyResourceFailures(newFailures)
+
+ byName := map[string]bool{} // name -> whether the mutator it was given
marks success
+ for _, u := range updater.updates {
+ cp := u.Mutator.Mutate(&apiv2.ApisixRoute{})
+ route := cp.(*apiv2.ApisixRoute)
+ accepted := false
+ for _, c := range route.Status.Conditions {
+ if c.Type == string(apiv2.ConditionTypeAccepted) {
+ accepted = c.Status == metav1.ConditionTrue
+ }
+ }
+ byName[u.NamespacedName.Name] = accepted
+ }
+
+ if accepted, ok := byName["resolved"]; !ok || !accepted {
+ t.Errorf("expected \"resolved\" to be written Accepted=true,
got present=%v accepted=%v", ok, accepted)
+ }
+ if accepted, ok := byName["still-bad"]; !ok || accepted {
+ t.Errorf("expected \"still-bad\" to be written Accepted=false,
got present=%v accepted=%v", ok, accepted)
+ }
+ if accepted, ok := byName["newly-bad"]; !ok || accepted {
+ t.Errorf("expected \"newly-bad\" to be written Accepted=false,
got present=%v accepted=%v", ok, accepted)
+ }
+ if _, ok := byName["resolved"]; len(byName) != 3 || !ok {
+ t.Errorf("expected exactly 3 writes (resolved, still-bad,
newly-bad), got %v", byName)
+ }
+
+ if len(d.resourceFailures) != 2 {
+ t.Errorf("d.resourceFailures should be replaced with
newFailures, got %v", d.resourceFailures)
+ }
+}
diff --git a/internal/provider/options.go b/internal/provider/options.go
index c47e7ce9..0aaf0e90 100644
--- a/internal/provider/options.go
+++ b/internal/provider/options.go
@@ -20,6 +20,9 @@ package provider
import (
"time"
+ "k8s.io/client-go/tools/record"
+ "sigs.k8s.io/controller-runtime/pkg/client"
+
"github.com/apache/apisix-ingress-controller/internal/controller/config"
)
@@ -34,6 +37,11 @@ type Options struct {
DefaultBackendMode string
DefaultResolveEndpoints bool
ListenerPortMatchMode config.ListenerPortMatchMode
+ EventRecorder record.EventRecorder
+ // K8sClient reads live Kubernetes objects, e.g. to fetch a resource's
real UID
+ // before recording an Event against it. Named to stay unambiguous next
to any
+ // provider-specific client, such as apisixProvider's own ADC client.
+ K8sClient client.Client
}
func (o *Options) ApplyToList(lo *Options) {
@@ -55,6 +63,12 @@ func (o *Options) ApplyToList(lo *Options) {
if o.ListenerPortMatchMode != "" {
lo.ListenerPortMatchMode = o.ListenerPortMatchMode
}
+ if o.EventRecorder != nil {
+ lo.EventRecorder = o.EventRecorder
+ }
+ if o.K8sClient != nil {
+ lo.K8sClient = o.K8sClient
+ }
}
func (o *Options) ApplyOptions(opts []Option) *Options {
diff --git a/test/e2e/crds/v2/route.go b/test/e2e/crds/v2/route.go
index 67b4a0b7..262550e8 100644
--- a/test/e2e/crds/v2/route.go
+++ b/test/e2e/crds/v2/route.go
@@ -2144,25 +2144,34 @@ spec:
err :=
s.CreateResourceFromString(fmt.Sprintf(apisixRouteSpec, s.Namespace()))
Expect(err).NotTo(HaveOccurred(), "creating
ApisixRoute")
- By("check ApisixRoute status")
+ // The data plane is entirely unreachable, ADC can't
even attempt a per-resource
+ // push, so there's nothing to attribute this to but
the GatewayProxy: see
+ // classifySyncResult.
+ By("check GatewayProxy status")
s.RetryAssertion(func() string {
- output, _ := s.GetOutputFromString("ar",
"default", "-o", "yaml", "-n", s.Namespace())
+ output, _ :=
s.GetOutputFromString("gatewayproxy", "apisix-proxy-config", "-o", "yaml",
"-n", s.Namespace())
return output
}).WithTimeout(30 * time.Second).
Should(
And(
+ ContainSubstring("type:
DataPlaneAvailable"),
ContainSubstring(`status:
"False"`),
- ContainSubstring(`reason:
SyncFailed`),
+ ContainSubstring("reason:
DataPlaneInstanceUnavailable"),
),
)
s.Deployer.ScaleDataplane(1)
s.RetryAssertion(func() string {
- output, _ := s.GetOutputFromString("ar",
"default", "-o", "yaml", "-n", s.Namespace())
+ output, _ :=
s.GetOutputFromString("gatewayproxy", "apisix-proxy-config", "-o", "yaml",
"-n", s.Namespace())
return output
}).WithTimeout(60 * time.Second).
- Should(ContainSubstring(`status: "True"`))
+ Should(
+ And(
+ ContainSubstring("type:
DataPlaneAvailable"),
+ ContainSubstring(`status:
"True"`),
+ ),
+ )
By("check route in APISIX")
s.RequestAssert(&scaffold.RequestAssert{
diff --git a/test/e2e/crds/v2/status.go b/test/e2e/crds/v2/status.go
index eaccb601..d43ddde7 100644
--- a/test/e2e/crds/v2/status.go
+++ b/test/e2e/crds/v2/status.go
@@ -109,7 +109,7 @@ spec:
servicePort: 80
`
It("unknown plugin", func() {
- if os.Getenv("PROVIDER_TYPE") == "apisix-standalone" {
+ if os.Getenv("PROVIDER_TYPE") ==
framework.ProviderTypeAPISIXStandalone {
Skip("apisix standalone does not validate
unknown plugins")
}
By("apply ApisixRoute with valid plugin")
@@ -170,15 +170,20 @@ spec:
err = s.CreateResourceFromString(string(newServiceYaml))
Expect(err).NotTo(HaveOccurred(), "creating service")
- By("check ApisixRoute status")
+ // This breaks the GatewayProxy's own admin API
address, not a route's backend,
+ // so ADC can't even attempt a per-resource push:
there's nothing to attribute
+ // this to but the GatewayProxy itself, for either
backend type (see
+ // classifySyncResult).
+ By("check GatewayProxy status")
s.RetryAssertion(func() string {
- output, _ := s.GetOutputFromString("ar",
"default", "-o", "yaml")
+ output, _ :=
s.GetOutputFromString("gatewayproxy", "apisix-proxy-config", "-o", "yaml")
return output
}).WithTimeout(60 * time.Second).
Should(
And(
+ ContainSubstring("type:
DataPlaneAvailable"),
ContainSubstring(`status:
"False"`),
- ContainSubstring(`reason:
SyncFailed`),
+ ContainSubstring("reason:
DataPlaneInstanceUnavailable"),
),
)
@@ -199,15 +204,16 @@ spec:
err = s.CreateResourceFromString(string(newServiceYaml))
Expect(err).NotTo(HaveOccurred(), "creating service")
- By("check ApisixRoute status after scaling up")
+ By("check GatewayProxy status after scaling up")
s.RetryAssertion(func() string {
- output, _ := s.GetOutputFromString("ar",
"default", "-o", "yaml")
+ output, _ :=
s.GetOutputFromString("gatewayproxy", "apisix-proxy-config", "-o", "yaml")
return output
}).WithTimeout(60 * time.Second).
Should(
And(
+ ContainSubstring("type:
DataPlaneAvailable"),
ContainSubstring(`status:
"True"`),
- ContainSubstring(`reason:
Accepted`),
+ ContainSubstring("reason:
DataPlaneAvailable"),
),
)
@@ -220,6 +226,116 @@ spec:
})
})
+ It("gateway proxy reports an unreachable data plane instance",
func() {
+ if os.Getenv("PROVIDER_TYPE") !=
framework.ProviderTypeAPISIXStandalone {
+ Skip("EndpointStatus, and the GatewayProxy
condition derived from it, only exists in apisix-standalone mode")
+ }
+
+ const name = "gateway-proxy-partial-instance"
+ gatewayProxyYaml := fmt.Sprintf(`
+apiVersion: apisix.apache.org/v1alpha1
+kind: GatewayProxy
+metadata:
+ name: %s
+ namespace: %s
+spec:
+ provider:
+ type: ControlPlane
+ controlPlane:
+ mode: apisix-standalone
+ endpoints:
+ - %s
+ - http://unreachable-instance.invalid:9180
+ - http://unreachable-instance-2.invalid:9180
+ auth:
+ type: AdminKey
+ adminKey:
+ value: "%s"
+`, name, s.Namespace(), s.Deployer.GetAdminEndpoint(), s.AdminKey())
+ By("create GatewayProxy with two unreachable endpoints")
+ err := s.CreateResourceFromString(gatewayProxyYaml)
+ Expect(err).NotTo(HaveOccurred(), "creating
GatewayProxy")
+
+ ingressClassYaml := fmt.Sprintf(`
+apiVersion: networking.k8s.io/v1
+kind: IngressClass
+metadata:
+ name: %s
+spec:
+ controller: %s
+ parameters:
+ apiGroup: "apisix.apache.org"
+ kind: "GatewayProxy"
+ name: %s
+ namespace: %s
+ scope: Namespace
+`, name, s.GetControllerName(), name, s.Namespace())
+ By("create IngressClass")
+ err =
s.CreateResourceFromStringWithNamespace(ingressClassYaml, "")
+ Expect(err).NotTo(HaveOccurred(), "creating
IngressClass")
+
+ By("apply ApisixRoute through it")
+ applier.MustApplyAPIv2(types.NamespacedName{Namespace:
s.Namespace(), Name: "default"}, &apiv2.ApisixRoute{}, fmt.Sprintf(ar,
s.Namespace(), name))
+
+ By("check route in APISIX")
+ s.RequestAssert(&scaffold.RequestAssert{
+ Method: "GET",
+ Path: "/get",
+ Headers: map[string]string{"Host": "httpbin"},
+ Check: scaffold.WithExpectedStatus(200),
+ })
+
+ By("check GatewayProxy status")
+ s.RetryAssertion(func() string {
+ output, _ :=
s.GetOutputFromString("gatewayproxy", name, "-o", "yaml", "-n", s.Namespace())
+ return output
+ }).Should(
+ And(
+ ContainSubstring("type:
DataPlaneAvailable"),
+ ContainSubstring(`status: "False"`),
+ ContainSubstring("reason:
DataPlaneInstanceUnavailable"),
+
ContainSubstring("unreachable-instance.invalid"),
+ ),
+ )
+
+ By("check a separate Warning event was recorded for
each unreachable endpoint")
+ s.RetryAssertion(func() []string {
+ output, err := s.GetOutputFromString("events",
+ "--field-selector",
"involvedObject.kind=GatewayProxy,involvedObject.name="+name,
+ "-n", s.Namespace(), "-o", "yaml")
+ if err != nil {
+ return nil
+ }
+ var events corev1.EventList
+ if err := yaml.Unmarshal([]byte(output),
&events); err != nil {
+ return nil
+ }
+ messages := make([]string, 0, len(events.Items))
+ for _, e := range events.Items {
+ if e.Type == "Warning" && e.Reason ==
"DataPlaneInstanceUnavailable" {
+ messages = append(messages,
e.Message)
+ }
+ }
+ return messages
+ }).Should(
+ ConsistOf(
+
ContainSubstring("unreachable-instance.invalid"),
+
ContainSubstring("unreachable-instance-2.invalid"),
+ ),
+ )
+
+ By("check the ApisixRoute itself is not marked as
failed")
+ s.RetryAssertion(func() string {
+ output, _ := s.GetOutputFromString("ar",
"default", "-o", "yaml", "-n", s.Namespace())
+ return output
+ }).Should(
+ And(
+ ContainSubstring(`status: "True"`),
+ ContainSubstring("reason: Accepted"),
+ ),
+ )
+ })
+
It("update the same status only once", func() {
By("apply ApisixRoute")
applier.MustApplyAPIv2(types.NamespacedName{Namespace:
s.Namespace(), Name: "default"}, &apiv2.ApisixRoute{}, fmt.Sprintf(ar,
s.Namespace(), s.Namespace()))
diff --git a/test/e2e/framework/manifests/ingress.yaml
b/test/e2e/framework/manifests/ingress.yaml
index 022bb159..99d34ed9 100644
--- a/test/e2e/framework/manifests/ingress.yaml
+++ b/test/e2e/framework/manifests/ingress.yaml
@@ -119,6 +119,7 @@ rules:
- apisixupstreams/status
- backendtrafficpolicies/status
- consumers/status
+ - gatewayproxies/status
- httproutepolicies/status
- l4routepolicies/status
verbs:
@@ -238,7 +239,7 @@ metadata:
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: Role
- name: {{ .Namespace }}-apisix-ingress-leader-election-role
+ name: apisix-ingress-leader-election-role
subjects:
- kind: ServiceAccount
name: apisix-ingress-controller-manager
diff --git a/test/e2e/gatewayapi/status.go b/test/e2e/gatewayapi/status.go
index a79d5559..9c4e2e3d 100644
--- a/test/e2e/gatewayapi/status.go
+++ b/test/e2e/gatewayapi/status.go
@@ -137,15 +137,20 @@ spec:
err = s.CreateResourceFromString(string(newServiceYaml))
Expect(err).NotTo(HaveOccurred(), "creating service")
- By("check ApisixRoute status")
+ // This breaks the GatewayProxy's own admin API
address, not a route's backend,
+ // so ADC can't even attempt a per-resource push:
there's nothing to attribute
+ // this to but the GatewayProxy itself, for either
backend type (see
+ // classifySyncResult).
+ By("check GatewayProxy status")
s.RetryAssertion(func() string {
- output, _ := s.GetOutputFromString("httproute",
"httpbin", "-o", "yaml")
+ output, _ :=
s.GetOutputFromString("gatewayproxy", "apisix-proxy-config", "-o", "yaml")
return output
}).WithTimeout(60 * time.Second).
Should(
And(
+ ContainSubstring("type:
DataPlaneAvailable"),
ContainSubstring(`status:
"False"`),
- ContainSubstring(`reason:
SyncFailed`),
+ ContainSubstring("reason:
DataPlaneInstanceUnavailable"),
),
)
