Compare commits
3 commits
f842abf71b
...
b7af30eb22
Author | SHA1 | Date | |
---|---|---|---|
b7af30eb22 | |||
|
51f7f610c9 | ||
|
5ed972ccd8 |
20 changed files with 18830 additions and 10704 deletions
2
.gitignore
vendored
2
.gitignore
vendored
|
@ -19,4 +19,4 @@ plugins-storage/
|
|||
plugins-local/
|
||||
traefik_changelog.md
|
||||
integration/tailscale.secret
|
||||
integration/conformance-reports/
|
||||
integration/conformance-reports/**/experimental-dev-default-report.yaml
|
||||
|
|
3
Makefile
3
Makefile
|
@ -103,7 +103,8 @@ test-integration: binary
|
|||
.PHONY: test-gateway-api-conformance
|
||||
#? test-gateway-api-conformance: Run the conformance tests
|
||||
test-gateway-api-conformance: build-image-dirty
|
||||
GOOS=$(GOOS) GOARCH=$(GOARCH) go test ./integration -v -test.run K8sConformanceSuite -k8sConformance $(TESTFLAGS)
|
||||
# In case of a new Minor/Major version, the k8sConformanceTraefikVersion needs to be updated.
|
||||
GOOS=$(GOOS) GOARCH=$(GOARCH) go test ./integration -v -test.run K8sConformanceSuite -k8sConformance -k8sConformanceTraefikVersion="v3.1" $(TESTFLAGS)
|
||||
|
||||
.PHONY: test-ui-unit
|
||||
#? test-ui-unit: Run the unit tests for the webui
|
||||
|
|
|
@ -91,6 +91,8 @@ You must run these local verifications before you submit your pull request to pr
|
|||
Your PR will not be reviewed until these are green on the CI.
|
||||
|
||||
* `make generate`
|
||||
* `make generate-crd`
|
||||
* `make test-gateway-api-conformance`
|
||||
* `make validate`
|
||||
* `make pull-images`
|
||||
* `make test`
|
||||
|
|
|
@ -75,3 +75,31 @@ To configure `kubernetesgateway`, please check out the [KubernetesGateway Provid
|
|||
|
||||
The Kubernetes Ingress provider option `disableIngressClassLookup` has been deprecated in v3.1.1, and will be removed in the next major version.
|
||||
Please use the `disableClusterScopeResources` option instead to avoid cluster scope resources discovery (IngressClass, Nodes).
|
||||
|
||||
## v3.1 to v3.2
|
||||
|
||||
### Kubernetes Gateway Provider RBACs
|
||||
|
||||
Starting with v3.2, the Kubernetes Gateway Provider now supports [GRPCRoute](https://gateway-api.sigs.k8s.io/api-types/grpcroute/).
|
||||
|
||||
Therefore, in the corresponding RBACs (see [KubernetesGateway](../reference/dynamic-configuration/kubernetes-gateway.md#rbac) provider RBACs),
|
||||
the `grcroutes` and `grpcroutes/status` rights have to be added.
|
||||
|
||||
```yaml
|
||||
...
|
||||
- apiGroups:
|
||||
- gateway.networking.k8s.io
|
||||
resources:
|
||||
- grpcroutes
|
||||
verbs:
|
||||
- get
|
||||
- list
|
||||
- watch
|
||||
- apiGroups:
|
||||
- gateway.networking.k8s.io
|
||||
resources:
|
||||
- grpcroutes/status
|
||||
verbs:
|
||||
- update
|
||||
...
|
||||
```
|
||||
|
|
|
@ -33,6 +33,7 @@ rules:
|
|||
- gatewayclasses
|
||||
- gateways
|
||||
- httproutes
|
||||
- grpcroutes
|
||||
- referencegrants
|
||||
- tcproutes
|
||||
- tlsroutes
|
||||
|
@ -46,6 +47,7 @@ rules:
|
|||
- gatewayclasses/status
|
||||
- gateways/status
|
||||
- httproutes/status
|
||||
- grpcroutes/status
|
||||
- tcproutes/status
|
||||
- tlsroutes/status
|
||||
verbs:
|
||||
|
|
|
@ -277,6 +277,158 @@ X-Forwarded-Server: traefik-6b66d45748-ns8mt
|
|||
X-Real-Ip: 10.42.1.0
|
||||
```
|
||||
|
||||
### GRPC
|
||||
|
||||
The `GRPCRoute` is an extended resource in the Gateway API specification, designed to define how GRPC traffic should be routed within a Kubernetes cluster.
|
||||
It allows the specification of routing rules that direct GRPC requests to the appropriate Kubernetes backend services.
|
||||
|
||||
For more details on the resource and concepts, check out the Kubernetes Gateway API [documentation](https://gateway-api.sigs.k8s.io/api-types/grpcroute/).
|
||||
|
||||
For example, the following manifests configure an echo backend and its corresponding `GRPCRoute`,
|
||||
reachable through the [deployed `Gateway`](#deploying-a-gateway) at the `echo.localhost:80` address.
|
||||
|
||||
```yaml tab="GRPCRoute"
|
||||
---
|
||||
apiVersion: gateway.networking.k8s.io/v1
|
||||
kind: GRPCRoute
|
||||
metadata:
|
||||
name: echo
|
||||
namespace: default
|
||||
spec:
|
||||
parentRefs:
|
||||
- name: traefik
|
||||
sectionName: http
|
||||
kind: Gateway
|
||||
|
||||
hostnames:
|
||||
- echo.localhost
|
||||
|
||||
rules:
|
||||
- matches:
|
||||
- method:
|
||||
type: Exact
|
||||
service: grpc.reflection.v1alpha.ServerReflection
|
||||
|
||||
- method:
|
||||
type: Exact
|
||||
service: gateway_api_conformance.echo_basic.grpcecho.GrpcEcho
|
||||
method: Echo
|
||||
|
||||
backendRefs:
|
||||
- name: echo
|
||||
namespace: default
|
||||
port: 3000
|
||||
```
|
||||
|
||||
```yaml tab="Echo deployment"
|
||||
---
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: echo
|
||||
namespace: default
|
||||
spec:
|
||||
selector:
|
||||
matchLabels:
|
||||
app: echo
|
||||
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app: echo
|
||||
spec:
|
||||
containers:
|
||||
- name: echo-basic
|
||||
image: gcr.io/k8s-staging-gateway-api/echo-basic
|
||||
env:
|
||||
- name: POD_NAME
|
||||
valueFrom:
|
||||
fieldRef:
|
||||
fieldPath: metadata.name
|
||||
- name: NAMESPACE
|
||||
valueFrom:
|
||||
fieldRef:
|
||||
fieldPath: metadata.namespace
|
||||
- name: GRPC_ECHO_SERVER
|
||||
value: "1"
|
||||
|
||||
---
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: echo
|
||||
namespace: default
|
||||
spec:
|
||||
selector:
|
||||
app: echo
|
||||
|
||||
ports:
|
||||
- port: 3000
|
||||
```
|
||||
|
||||
Once everything is deployed, sending a GRPC request to the HTTP endpoint should return the following responses:
|
||||
|
||||
```shell
|
||||
$ grpcurl -plaintext echo.localhost:80 gateway_api_conformance.echo_basic.grpcecho.GrpcEcho/Echo
|
||||
|
||||
{
|
||||
"assertions": {
|
||||
"fullyQualifiedMethod": "/gateway_api_conformance.echo_basic.grpcecho.GrpcEcho/Echo",
|
||||
"headers": [
|
||||
{
|
||||
"key": "x-real-ip",
|
||||
"value": "10.42.2.0"
|
||||
},
|
||||
{
|
||||
"key": "x-forwarded-server",
|
||||
"value": "traefik-74b4cf85d8-nkqqf"
|
||||
},
|
||||
{
|
||||
"key": "x-forwarded-port",
|
||||
"value": "80"
|
||||
},
|
||||
{
|
||||
"key": "x-forwarded-for",
|
||||
"value": "10.42.2.0"
|
||||
},
|
||||
{
|
||||
"key": "grpc-accept-encoding",
|
||||
"value": "gzip"
|
||||
},
|
||||
{
|
||||
"key": "user-agent",
|
||||
"value": "grpcurl/1.9.1 grpc-go/1.61.0"
|
||||
},
|
||||
{
|
||||
"key": "content-type",
|
||||
"value": "application/grpc"
|
||||
},
|
||||
{
|
||||
"key": "x-forwarded-host",
|
||||
"value": "echo.localhost:80"
|
||||
},
|
||||
{
|
||||
"key": ":authority",
|
||||
"value": "echo.localhost:80"
|
||||
},
|
||||
{
|
||||
"key": "accept-encoding",
|
||||
"value": "gzip"
|
||||
},
|
||||
{
|
||||
"key": "x-forwarded-proto",
|
||||
"value": "http"
|
||||
}
|
||||
],
|
||||
"authority": "echo.localhost:80",
|
||||
"context": {
|
||||
"namespace": "default",
|
||||
"pod": "echo-78f76675cf-9k7rf"
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### TCP
|
||||
|
||||
!!! info "Experimental Channel"
|
||||
|
|
|
@ -0,0 +1,55 @@
|
|||
apiVersion: gateway.networking.k8s.io/v1alpha1
|
||||
date: '-'
|
||||
gatewayAPIChannel: experimental
|
||||
gatewayAPIVersion: v1.1.0
|
||||
implementation:
|
||||
contact:
|
||||
- '@traefik/maintainers'
|
||||
organization: traefik
|
||||
project: traefik
|
||||
url: https://traefik.io/
|
||||
version: v3.1
|
||||
kind: ConformanceReport
|
||||
mode: default
|
||||
profiles:
|
||||
- core:
|
||||
result: success
|
||||
statistics:
|
||||
Failed: 0
|
||||
Passed: 12
|
||||
Skipped: 0
|
||||
name: GATEWAY-GRPC
|
||||
summary: Core tests succeeded.
|
||||
- core:
|
||||
result: success
|
||||
statistics:
|
||||
Failed: 0
|
||||
Passed: 33
|
||||
Skipped: 0
|
||||
extended:
|
||||
result: success
|
||||
statistics:
|
||||
Failed: 0
|
||||
Passed: 10
|
||||
Skipped: 0
|
||||
supportedFeatures:
|
||||
- GatewayPort8080
|
||||
- HTTPRouteHostRewrite
|
||||
- HTTPRouteMethodMatching
|
||||
- HTTPRoutePathRedirect
|
||||
- HTTPRoutePathRewrite
|
||||
- HTTPRoutePortRedirect
|
||||
- HTTPRouteQueryParamMatching
|
||||
- HTTPRouteResponseHeaderModification
|
||||
- HTTPRouteSchemeRedirect
|
||||
unsupportedFeatures:
|
||||
- GatewayHTTPListenerIsolation
|
||||
- GatewayStaticAddresses
|
||||
- HTTPRouteBackendRequestHeaderModification
|
||||
- HTTPRouteBackendTimeout
|
||||
- HTTPRouteParentRefPort
|
||||
- HTTPRouteRequestMirror
|
||||
- HTTPRouteRequestMultipleMirrors
|
||||
- HTTPRouteRequestTimeout
|
||||
name: GATEWAY-HTTP
|
||||
summary: Core tests succeeded. Extended tests succeeded.
|
|
@ -34,6 +34,7 @@ rules:
|
|||
- gatewayclasses
|
||||
- gateways
|
||||
- httproutes
|
||||
- grpcroutes
|
||||
- tcproutes
|
||||
- tlsroutes
|
||||
- referencegrants
|
||||
|
@ -47,6 +48,7 @@ rules:
|
|||
- gatewayclasses/status
|
||||
- gateways/status
|
||||
- httproutes/status
|
||||
- grpcroutes/status
|
||||
- tcproutes/status
|
||||
- tlsroutes/status
|
||||
- referencegrants/status
|
||||
|
|
File diff suppressed because it is too large
Load diff
|
@ -39,6 +39,7 @@ var (
|
|||
showLog = flag.Bool("tlog", false, "always show Traefik logs")
|
||||
k8sConformance = flag.Bool("k8sConformance", false, "run K8s Gateway API conformance test")
|
||||
k8sConformanceRunTest = flag.String("k8sConformanceRunTest", "", "run a specific K8s Gateway API conformance test")
|
||||
k8sConformanceTraefikVersion = flag.String("k8sConformanceTraefikVersion", "dev", "specify the Traefik version for the K8s Gateway API conformance report")
|
||||
)
|
||||
|
||||
const tailscaleSecretFilePath = "tailscale.secret"
|
||||
|
|
|
@ -17,7 +17,6 @@ import (
|
|||
"github.com/testcontainers/testcontainers-go/modules/k3s"
|
||||
"github.com/testcontainers/testcontainers-go/network"
|
||||
"github.com/traefik/traefik/v3/integration/try"
|
||||
"github.com/traefik/traefik/v3/pkg/version"
|
||||
apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
kclientset "k8s.io/client-go/kubernetes"
|
||||
|
@ -191,13 +190,14 @@ func (s *K8sConformanceSuite) TestK8sGatewayAPIConformance() {
|
|||
Organization: "traefik",
|
||||
Project: "traefik",
|
||||
URL: "https://traefik.io/",
|
||||
Version: version.Version,
|
||||
Version: *k8sConformanceTraefikVersion,
|
||||
Contact: []string{"@traefik/maintainers"},
|
||||
},
|
||||
ConformanceProfiles: sets.New(ksuite.GatewayHTTPConformanceProfileName),
|
||||
ConformanceProfiles: sets.New(ksuite.GatewayHTTPConformanceProfileName, ksuite.GatewayGRPCConformanceProfileName),
|
||||
SupportedFeatures: sets.New(
|
||||
features.SupportGateway,
|
||||
features.SupportGatewayPort8080,
|
||||
features.SupportGRPCRoute,
|
||||
features.SupportHTTPRoute,
|
||||
features.SupportHTTPRouteQueryParamMatching,
|
||||
features.SupportHTTPRouteMethodMatching,
|
||||
|
@ -219,12 +219,17 @@ func (s *K8sConformanceSuite) TestK8sGatewayAPIConformance() {
|
|||
report, err := cSuite.Report()
|
||||
require.NoError(s.T(), err, "failed generating conformance report")
|
||||
|
||||
// Ignore report date to avoid diff with CI job.
|
||||
// However, we can track the date of the report thanks to the commit.
|
||||
// TODO: to publish this report automatically, we have to figure out how to handle the date diff.
|
||||
report.Date = "-"
|
||||
|
||||
rawReport, err := yaml.Marshal(report)
|
||||
require.NoError(s.T(), err)
|
||||
s.T().Logf("Conformance report:\n%s", string(rawReport))
|
||||
|
||||
require.NoError(s.T(), os.MkdirAll("./conformance-reports", 0o755))
|
||||
outFile := filepath.Join("conformance-reports", fmt.Sprintf("%s-%s-%s-report.yaml", report.GatewayAPIChannel, report.Version, report.Mode))
|
||||
require.NoError(s.T(), os.MkdirAll("./conformance-reports/"+report.GatewayAPIVersion, 0o755))
|
||||
outFile := filepath.Join("conformance-reports/"+report.GatewayAPIVersion, fmt.Sprintf("%s-%s-%s-report.yaml", report.GatewayAPIChannel, report.Version, report.Mode))
|
||||
require.NoError(s.T(), os.WriteFile(outFile, rawReport, 0o600))
|
||||
s.T().Logf("Report written to: %s", outFile)
|
||||
}
|
||||
|
|
|
@ -7,6 +7,7 @@ import (
|
|||
ptypes "github.com/traefik/paerser/types"
|
||||
traefiktls "github.com/traefik/traefik/v3/pkg/tls"
|
||||
"github.com/traefik/traefik/v3/pkg/types"
|
||||
"google.golang.org/grpc/codes"
|
||||
)
|
||||
|
||||
const (
|
||||
|
@ -132,6 +133,9 @@ type WRRService struct {
|
|||
// Status defines an HTTP status code that should be returned when calling the service.
|
||||
// This is required by the Gateway API implementation which expects specific HTTP status to be returned.
|
||||
Status *int `json:"-" toml:"-" yaml:"-" label:"-" file:"-"`
|
||||
// GRPCStatus defines a GRPC status code that should be returned when calling the service.
|
||||
// This is required by the Gateway API implementation which expects specific GRPC status to be returned.
|
||||
GRPCStatus *GRPCStatus `json:"-" toml:"-" yaml:"-" label:"-" file:"-"`
|
||||
}
|
||||
|
||||
// SetDefaults Default values for a WRRService.
|
||||
|
@ -142,6 +146,13 @@ func (w *WRRService) SetDefaults() {
|
|||
|
||||
// +k8s:deepcopy-gen=true
|
||||
|
||||
type GRPCStatus struct {
|
||||
Code codes.Code `json:"code,omitempty" toml:"code,omitempty" yaml:"code,omitempty" export:"true"`
|
||||
Msg string `json:"msg,omitempty" toml:"msg,omitempty" yaml:"msg,omitempty" export:"true"`
|
||||
}
|
||||
|
||||
// +k8s:deepcopy-gen=true
|
||||
|
||||
// Sticky holds the sticky configuration.
|
||||
type Sticky struct {
|
||||
// Cookie defines the sticky cookie configuration.
|
||||
|
|
|
@ -394,6 +394,22 @@ func (in *ForwardingTimeouts) DeepCopy() *ForwardingTimeouts {
|
|||
return out
|
||||
}
|
||||
|
||||
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
|
||||
func (in *GRPCStatus) DeepCopyInto(out *GRPCStatus) {
|
||||
*out = *in
|
||||
return
|
||||
}
|
||||
|
||||
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new GRPCStatus.
|
||||
func (in *GRPCStatus) DeepCopy() *GRPCStatus {
|
||||
if in == nil {
|
||||
return nil
|
||||
}
|
||||
out := new(GRPCStatus)
|
||||
in.DeepCopyInto(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
|
||||
func (in *GrpcWeb) DeepCopyInto(out *GrpcWeb) {
|
||||
*out = *in
|
||||
|
@ -2286,6 +2302,11 @@ func (in *WRRService) DeepCopyInto(out *WRRService) {
|
|||
*out = new(int)
|
||||
**out = **in
|
||||
}
|
||||
if in.GRPCStatus != nil {
|
||||
in, out := &in.GRPCStatus, &out.GRPCStatus
|
||||
*out = new(GRPCStatus)
|
||||
**out = **in
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
@ -56,11 +56,13 @@ type Client interface {
|
|||
UpdateGatewayStatus(ctx context.Context, gateway ktypes.NamespacedName, status gatev1.GatewayStatus) error
|
||||
UpdateGatewayClassStatus(ctx context.Context, name string, status gatev1.GatewayClassStatus) error
|
||||
UpdateHTTPRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1.HTTPRouteStatus) error
|
||||
UpdateGRPCRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1.GRPCRouteStatus) error
|
||||
UpdateTCPRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1alpha2.TCPRouteStatus) error
|
||||
UpdateTLSRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1alpha2.TLSRouteStatus) error
|
||||
ListGatewayClasses() ([]*gatev1.GatewayClass, error)
|
||||
ListGateways() []*gatev1.Gateway
|
||||
ListHTTPRoutes() ([]*gatev1.HTTPRoute, error)
|
||||
ListGRPCRoutes() ([]*gatev1.GRPCRoute, error)
|
||||
ListTCPRoutes() ([]*gatev1alpha2.TCPRoute, error)
|
||||
ListTLSRoutes() ([]*gatev1alpha2.TLSRoute, error)
|
||||
ListNamespaces(selector labels.Selector) ([]string, error)
|
||||
|
@ -205,6 +207,10 @@ func (c *clientWrapper) WatchAll(namespaces []string, stopCh <-chan struct{}) (<
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, err = factoryGateway.Gateway().V1().GRPCRoutes().Informer().AddEventHandler(eventHandler)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, err = factoryGateway.Gateway().V1beta1().ReferenceGrants().Informer().AddEventHandler(eventHandler)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
@ -317,6 +323,20 @@ func (c *clientWrapper) ListHTTPRoutes() ([]*gatev1.HTTPRoute, error) {
|
|||
return httpRoutes, nil
|
||||
}
|
||||
|
||||
func (c *clientWrapper) ListGRPCRoutes() ([]*gatev1.GRPCRoute, error) {
|
||||
var grpcRoutes []*gatev1.GRPCRoute
|
||||
for _, namespace := range c.watchedNamespaces {
|
||||
routes, err := c.factoriesGateway[c.lookupNamespace(namespace)].Gateway().V1().GRPCRoutes().Lister().GRPCRoutes(namespace).List(labels.Everything())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("listing GRPC routes in namespace %s", namespace)
|
||||
}
|
||||
|
||||
grpcRoutes = append(grpcRoutes, routes...)
|
||||
}
|
||||
|
||||
return grpcRoutes, nil
|
||||
}
|
||||
|
||||
func (c *clientWrapper) ListTCPRoutes() ([]*gatev1alpha2.TCPRoute, error) {
|
||||
var tcpRoutes []*gatev1alpha2.TCPRoute
|
||||
for _, namespace := range c.watchedNamespaces {
|
||||
|
@ -497,6 +517,58 @@ func (c *clientWrapper) UpdateHTTPRouteStatus(ctx context.Context, route ktypes.
|
|||
return nil
|
||||
}
|
||||
|
||||
func (c *clientWrapper) UpdateGRPCRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1.GRPCRouteStatus) error {
|
||||
if !c.isWatchedNamespace(route.Namespace) {
|
||||
return fmt.Errorf("updating GRPCRoute status %s/%s: namespace is not within watched namespaces", route.Namespace, route.Name)
|
||||
}
|
||||
|
||||
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
||||
currentRoute, err := c.factoriesGateway[c.lookupNamespace(route.Namespace)].Gateway().V1().GRPCRoutes().Lister().GRPCRoutes(route.Namespace).Get(route.Name)
|
||||
if err != nil {
|
||||
// We have to return err itself here (not wrapped inside another error)
|
||||
// so that RetryOnConflict can identify it correctly.
|
||||
return err
|
||||
}
|
||||
|
||||
parentStatuses := make([]gatev1.RouteParentStatus, len(status.Parents))
|
||||
copy(parentStatuses, status.Parents)
|
||||
|
||||
// keep statuses added by other gateway controllers.
|
||||
// TODO: we should also keep statuses for gateways managed by other Traefik instances.
|
||||
for _, parentStatus := range currentRoute.Status.Parents {
|
||||
if parentStatus.ControllerName != controllerName {
|
||||
parentStatuses = append(parentStatuses, parentStatus)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// do not update status when nothing has changed.
|
||||
if routeParentStatusesEqual(currentRoute.Status.Parents, parentStatuses) {
|
||||
return nil
|
||||
}
|
||||
|
||||
currentRoute = currentRoute.DeepCopy()
|
||||
currentRoute.Status = gatev1.GRPCRouteStatus{
|
||||
RouteStatus: gatev1.RouteStatus{
|
||||
Parents: parentStatuses,
|
||||
},
|
||||
}
|
||||
|
||||
if _, err = c.csGateway.GatewayV1().GRPCRoutes(route.Namespace).UpdateStatus(ctx, currentRoute, metav1.UpdateOptions{}); err != nil {
|
||||
// We have to return err itself here (not wrapped inside another error)
|
||||
// so that RetryOnConflict can identify it correctly.
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to update GRPCRoute %q status: %w", route.Name, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *clientWrapper) UpdateTCPRouteStatus(ctx context.Context, route ktypes.NamespacedName, status gatev1alpha2.TCPRouteStatus) error {
|
||||
if !c.isWatchedNamespace(route.Namespace) {
|
||||
return fmt.Errorf("updating TCPRoute status %s/%s: namespace is not within watched namespaces", route.Namespace, route.Name)
|
||||
|
|
439
pkg/provider/kubernetes/gateway/grpcroute.go
Normal file
439
pkg/provider/kubernetes/gateway/grpcroute.go
Normal file
|
@ -0,0 +1,439 @@
|
|||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
"github.com/traefik/traefik/v3/pkg/config/dynamic"
|
||||
"github.com/traefik/traefik/v3/pkg/provider"
|
||||
"google.golang.org/grpc/codes"
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
ktypes "k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/utils/ptr"
|
||||
gatev1 "sigs.k8s.io/gateway-api/apis/v1"
|
||||
)
|
||||
|
||||
// TODO: as described in the specification https://gateway-api.sigs.k8s.io/reference/spec/#gateway.networking.k8s.io%2fv1.GRPCRoute, we should check for hostname conflicts between HTTP and GRPC routes.
|
||||
func (p *Provider) loadGRPCRoutes(ctx context.Context, gatewayListeners []gatewayListener, conf *dynamic.Configuration) {
|
||||
routes, err := p.client.ListGRPCRoutes()
|
||||
if err != nil {
|
||||
log.Ctx(ctx).Error().Err(err).Msg("Unable to list GRPCRoutes")
|
||||
return
|
||||
}
|
||||
|
||||
for _, route := range routes {
|
||||
logger := log.Ctx(ctx).With().
|
||||
Str("grpc_route", route.Name).
|
||||
Str("namespace", route.Namespace).
|
||||
Logger()
|
||||
|
||||
var parentStatuses []gatev1.RouteParentStatus
|
||||
for _, parentRef := range route.Spec.ParentRefs {
|
||||
parentStatus := &gatev1.RouteParentStatus{
|
||||
ParentRef: parentRef,
|
||||
ControllerName: controllerName,
|
||||
Conditions: []metav1.Condition{
|
||||
{
|
||||
Type: string(gatev1.RouteConditionAccepted),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonNoMatchingParent),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, listener := range gatewayListeners {
|
||||
if !matchListener(listener, route.Namespace, parentRef) {
|
||||
continue
|
||||
}
|
||||
|
||||
accepted := true
|
||||
if !allowRoute(listener, route.Namespace, kindGRPCRoute) {
|
||||
parentStatus.Conditions = updateRouteConditionAccepted(parentStatus.Conditions, string(gatev1.RouteReasonNotAllowedByListeners))
|
||||
accepted = false
|
||||
}
|
||||
hostnames, ok := findMatchingHostnames(listener.Hostname, route.Spec.Hostnames)
|
||||
if !ok {
|
||||
parentStatus.Conditions = updateRouteConditionAccepted(parentStatus.Conditions, string(gatev1.RouteReasonNoMatchingListenerHostname))
|
||||
accepted = false
|
||||
}
|
||||
|
||||
if accepted {
|
||||
// Gateway listener should have AttachedRoutes set even when Gateway has unresolved refs.
|
||||
listener.Status.AttachedRoutes++
|
||||
// Only consider the route attached if the listener is in an "attached" state.
|
||||
if listener.Attached {
|
||||
parentStatus.Conditions = updateRouteConditionAccepted(parentStatus.Conditions, string(gatev1.RouteReasonAccepted))
|
||||
}
|
||||
}
|
||||
|
||||
routeConf, resolveRefCondition := p.loadGRPCRoute(logger.WithContext(ctx), listener, route, hostnames)
|
||||
if accepted && listener.Attached {
|
||||
mergeHTTPConfiguration(routeConf, conf)
|
||||
}
|
||||
|
||||
parentStatus.Conditions = upsertRouteConditionResolvedRefs(parentStatus.Conditions, resolveRefCondition)
|
||||
}
|
||||
|
||||
parentStatuses = append(parentStatuses, *parentStatus)
|
||||
}
|
||||
|
||||
status := gatev1.GRPCRouteStatus{
|
||||
RouteStatus: gatev1.RouteStatus{
|
||||
Parents: parentStatuses,
|
||||
},
|
||||
}
|
||||
if err := p.client.UpdateGRPCRouteStatus(ctx, ktypes.NamespacedName{Namespace: route.Namespace, Name: route.Name}, status); err != nil {
|
||||
logger.Warn().
|
||||
Err(err).
|
||||
Msg("Unable to update GRPCRoute status")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Provider) loadGRPCRoute(ctx context.Context, listener gatewayListener, route *gatev1.GRPCRoute, hostnames []gatev1.Hostname) (*dynamic.Configuration, metav1.Condition) {
|
||||
conf := &dynamic.Configuration{
|
||||
HTTP: &dynamic.HTTPConfiguration{
|
||||
Routers: make(map[string]*dynamic.Router),
|
||||
Middlewares: make(map[string]*dynamic.Middleware),
|
||||
Services: make(map[string]*dynamic.Service),
|
||||
ServersTransports: make(map[string]*dynamic.ServersTransport),
|
||||
},
|
||||
}
|
||||
|
||||
condition := metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionTrue,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteConditionResolvedRefs),
|
||||
}
|
||||
|
||||
for ri, routeRule := range route.Spec.Rules {
|
||||
// Adding the gateway desc and the entryPoint desc prevents overlapping of routers build from the same routes.
|
||||
routeKey := provider.Normalize(fmt.Sprintf("%s-%s-%s-%s-%d", route.Namespace, route.Name, listener.GWName, listener.EPName, ri))
|
||||
|
||||
matches := routeRule.Matches
|
||||
if len(matches) == 0 {
|
||||
matches = []gatev1.GRPCRouteMatch{{}}
|
||||
}
|
||||
|
||||
for _, match := range matches {
|
||||
rule, priority := buildGRPCMatchRule(hostnames, match)
|
||||
|
||||
router := dynamic.Router{
|
||||
RuleSyntax: "v3",
|
||||
Rule: rule,
|
||||
Priority: priority,
|
||||
EntryPoints: []string{listener.EPName},
|
||||
}
|
||||
if listener.Protocol == gatev1.HTTPSProtocolType {
|
||||
router.TLS = &dynamic.RouterTLSConfig{}
|
||||
}
|
||||
|
||||
var err error
|
||||
routerName := makeRouterName(rule, routeKey)
|
||||
router.Middlewares, err = p.loadGRPCMiddlewares(conf, route.Namespace, routerName, routeRule.Filters)
|
||||
switch {
|
||||
case err != nil:
|
||||
log.Ctx(ctx).Error().Err(err).Msg("Unable to load GRPC route filters")
|
||||
|
||||
errWrrName := routerName + "-err-wrr"
|
||||
conf.HTTP.Services[errWrrName] = &dynamic.Service{
|
||||
Weighted: &dynamic.WeightedRoundRobin{
|
||||
Services: []dynamic.WRRService{
|
||||
{
|
||||
Name: "invalid-grpcroute-filter",
|
||||
GRPCStatus: &dynamic.GRPCStatus{
|
||||
Code: codes.Unavailable,
|
||||
Msg: "Service Unavailable",
|
||||
},
|
||||
Weight: ptr.To(1),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
router.Service = errWrrName
|
||||
|
||||
default:
|
||||
var serviceCondition *metav1.Condition
|
||||
router.Service, serviceCondition = p.loadGRPCService(conf, routeKey, routeRule, route)
|
||||
if serviceCondition != nil {
|
||||
condition = *serviceCondition
|
||||
}
|
||||
}
|
||||
|
||||
conf.HTTP.Routers[routerName] = &router
|
||||
}
|
||||
}
|
||||
|
||||
return conf, condition
|
||||
}
|
||||
|
||||
func (p *Provider) loadGRPCService(conf *dynamic.Configuration, routeKey string, routeRule gatev1.GRPCRouteRule, route *gatev1.GRPCRoute) (string, *metav1.Condition) {
|
||||
name := routeKey + "-wrr"
|
||||
if _, ok := conf.HTTP.Services[name]; ok {
|
||||
return name, nil
|
||||
}
|
||||
|
||||
var wrr dynamic.WeightedRoundRobin
|
||||
var condition *metav1.Condition
|
||||
for _, backendRef := range routeRule.BackendRefs {
|
||||
svcName, svc, errCondition := p.loadGRPCBackendRef(route, backendRef)
|
||||
weight := ptr.To(int(ptr.Deref(backendRef.Weight, 1)))
|
||||
if errCondition != nil {
|
||||
condition = errCondition
|
||||
wrr.Services = append(wrr.Services, dynamic.WRRService{
|
||||
Name: svcName,
|
||||
GRPCStatus: &dynamic.GRPCStatus{
|
||||
Code: codes.Unavailable,
|
||||
Msg: "Service Unavailable",
|
||||
},
|
||||
Weight: weight,
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
if svc != nil {
|
||||
conf.HTTP.Services[svcName] = svc
|
||||
}
|
||||
|
||||
wrr.Services = append(wrr.Services, dynamic.WRRService{
|
||||
Name: svcName,
|
||||
Weight: weight,
|
||||
})
|
||||
}
|
||||
|
||||
conf.HTTP.Services[name] = &dynamic.Service{Weighted: &wrr}
|
||||
return name, condition
|
||||
}
|
||||
|
||||
func (p *Provider) loadGRPCBackendRef(route *gatev1.GRPCRoute, backendRef gatev1.GRPCBackendRef) (string, *dynamic.Service, *metav1.Condition) {
|
||||
kind := ptr.Deref(backendRef.Kind, "Service")
|
||||
|
||||
group := groupCore
|
||||
if backendRef.Group != nil && *backendRef.Group != "" {
|
||||
group = string(*backendRef.Group)
|
||||
}
|
||||
|
||||
namespace := route.Namespace
|
||||
if backendRef.Namespace != nil && *backendRef.Namespace != "" {
|
||||
namespace = string(*backendRef.Namespace)
|
||||
}
|
||||
|
||||
serviceName := provider.Normalize(namespace + "-" + string(backendRef.Name))
|
||||
|
||||
if group != groupCore || kind != "Service" {
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonInvalidKind),
|
||||
Message: fmt.Sprintf("Cannot load GRPCBackendRef %s/%s/%s/%s: only Kubernetes services are supported", group, kind, namespace, backendRef.Name),
|
||||
}
|
||||
}
|
||||
|
||||
if err := p.isReferenceGranted(groupGateway, kindGRPCRoute, route.Namespace, group, string(kind), string(backendRef.Name), namespace); err != nil {
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonRefNotPermitted),
|
||||
Message: fmt.Sprintf("Cannot load GRPCBackendRef %s/%s/%s/%s: %s", group, kind, namespace, backendRef.Name, err),
|
||||
}
|
||||
}
|
||||
|
||||
port := ptr.Deref(backendRef.Port, gatev1.PortNumber(0))
|
||||
if port == 0 {
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonUnsupportedProtocol),
|
||||
Message: fmt.Sprintf("Cannot load GRPCBackendRef %s/%s/%s/%s port is required", group, kind, namespace, backendRef.Name),
|
||||
}
|
||||
}
|
||||
|
||||
portStr := strconv.FormatInt(int64(port), 10)
|
||||
serviceName = provider.Normalize(serviceName + "-" + portStr)
|
||||
|
||||
lb, err := p.loadGRPCServers(namespace, backendRef)
|
||||
if err != nil {
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonBackendNotFound),
|
||||
Message: fmt.Sprintf("Cannot load GRPCBackendRef %s/%s/%s/%s: %s", group, kind, namespace, backendRef.Name, err),
|
||||
}
|
||||
}
|
||||
|
||||
return serviceName, &dynamic.Service{LoadBalancer: lb}, nil
|
||||
}
|
||||
|
||||
func (p *Provider) loadGRPCMiddlewares(conf *dynamic.Configuration, namespace, routerName string, filters []gatev1.GRPCRouteFilter) ([]string, error) {
|
||||
middlewares := make(map[string]*dynamic.Middleware)
|
||||
for i, filter := range filters {
|
||||
name := fmt.Sprintf("%s-%s-%d", routerName, strings.ToLower(string(filter.Type)), i)
|
||||
switch filter.Type {
|
||||
case gatev1.GRPCRouteFilterRequestHeaderModifier:
|
||||
middlewares[name] = createRequestHeaderModifier(filter.RequestHeaderModifier)
|
||||
|
||||
case gatev1.GRPCRouteFilterExtensionRef:
|
||||
name, middleware, err := p.loadHTTPRouteFilterExtensionRef(namespace, filter.ExtensionRef)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loading ExtensionRef filter %s: %w", filter.Type, err)
|
||||
}
|
||||
|
||||
middlewares[name] = middleware
|
||||
|
||||
default:
|
||||
// As per the spec: https://gateway-api.sigs.k8s.io/api-types/httproute/#filters-optional
|
||||
// In all cases where incompatible or unsupported filters are
|
||||
// specified, implementations MUST add a warning condition to
|
||||
// status.
|
||||
return nil, fmt.Errorf("unsupported filter %s", filter.Type)
|
||||
}
|
||||
}
|
||||
|
||||
var middlewareNames []string
|
||||
for name, middleware := range middlewares {
|
||||
if middleware != nil {
|
||||
conf.HTTP.Middlewares[name] = middleware
|
||||
}
|
||||
|
||||
middlewareNames = append(middlewareNames, name)
|
||||
}
|
||||
|
||||
return middlewareNames, nil
|
||||
}
|
||||
|
||||
func (p *Provider) loadGRPCServers(namespace string, backendRef gatev1.GRPCBackendRef) (*dynamic.ServersLoadBalancer, error) {
|
||||
if backendRef.Port == nil {
|
||||
return nil, errors.New("port is required for Kubernetes Service reference")
|
||||
}
|
||||
|
||||
service, exists, err := p.client.GetService(namespace, string(backendRef.Name))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("getting service: %w", err)
|
||||
}
|
||||
if !exists {
|
||||
return nil, errors.New("service not found")
|
||||
}
|
||||
|
||||
var svcPort *corev1.ServicePort
|
||||
for _, p := range service.Spec.Ports {
|
||||
if p.Port == int32(*backendRef.Port) {
|
||||
svcPort = &p
|
||||
break
|
||||
}
|
||||
}
|
||||
if svcPort == nil {
|
||||
return nil, fmt.Errorf("service port %d not found", *backendRef.Port)
|
||||
}
|
||||
|
||||
endpointSlices, err := p.client.ListEndpointSlicesForService(namespace, string(backendRef.Name))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("getting endpointslices: %w", err)
|
||||
}
|
||||
if len(endpointSlices) == 0 {
|
||||
return nil, errors.New("endpointslices not found")
|
||||
}
|
||||
|
||||
lb := &dynamic.ServersLoadBalancer{}
|
||||
lb.SetDefaults()
|
||||
|
||||
addresses := map[string]struct{}{}
|
||||
for _, endpointSlice := range endpointSlices {
|
||||
var port int32
|
||||
for _, p := range endpointSlice.Ports {
|
||||
if svcPort.Name == *p.Name {
|
||||
port = *p.Port
|
||||
break
|
||||
}
|
||||
}
|
||||
if port == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, endpoint := range endpointSlice.Endpoints {
|
||||
if endpoint.Conditions.Ready == nil || !*endpoint.Conditions.Ready {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, address := range endpoint.Addresses {
|
||||
if _, ok := addresses[address]; ok {
|
||||
continue
|
||||
}
|
||||
|
||||
addresses[address] = struct{}{}
|
||||
lb.Servers = append(lb.Servers, dynamic.Server{
|
||||
URL: fmt.Sprintf("h2c://%s", net.JoinHostPort(address, strconv.Itoa(int(port)))),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return lb, nil
|
||||
}
|
||||
|
||||
func buildGRPCMatchRule(hostnames []gatev1.Hostname, match gatev1.GRPCRouteMatch) (string, int) {
|
||||
var matchRules []string
|
||||
|
||||
methodRule := buildGRPCMethodRule(match.Method)
|
||||
matchRules = append(matchRules, methodRule)
|
||||
|
||||
headerRules := buildGRPCHeaderRules(match.Headers)
|
||||
matchRules = append(matchRules, headerRules...)
|
||||
|
||||
matchRulesStr := strings.Join(matchRules, " && ")
|
||||
|
||||
hostRule, priority := buildHostRule(hostnames)
|
||||
if hostRule == "" {
|
||||
return matchRulesStr, len(matchRulesStr)
|
||||
}
|
||||
return hostRule + " && " + matchRulesStr, priority + len(matchRulesStr)
|
||||
}
|
||||
|
||||
func buildGRPCMethodRule(method *gatev1.GRPCMethodMatch) string {
|
||||
if method == nil {
|
||||
return "PathPrefix(`/`)"
|
||||
}
|
||||
|
||||
sExpr := "[^/]+"
|
||||
if s := ptr.Deref(method.Service, ""); s != "" {
|
||||
sExpr = s
|
||||
}
|
||||
|
||||
mExpr := "[^/]+"
|
||||
if m := ptr.Deref(method.Method, ""); m != "" {
|
||||
mExpr = m
|
||||
}
|
||||
|
||||
return fmt.Sprintf("PathRegexp(`/%s/%s`)", sExpr, mExpr)
|
||||
}
|
||||
|
||||
func buildGRPCHeaderRules(headers []gatev1.GRPCHeaderMatch) []string {
|
||||
var rules []string
|
||||
for _, header := range headers {
|
||||
switch ptr.Deref(header.Type, gatev1.HeaderMatchExact) {
|
||||
case gatev1.HeaderMatchExact:
|
||||
rules = append(rules, fmt.Sprintf("Header(`%s`,`%s`)", header.Name, header.Value))
|
||||
case gatev1.HeaderMatchRegularExpression:
|
||||
rules = append(rules, fmt.Sprintf("HeaderRegexp(`%s`,`%s`)", header.Name, header.Value))
|
||||
}
|
||||
}
|
||||
|
||||
return rules
|
||||
}
|
227
pkg/provider/kubernetes/gateway/grpcroute_test.go
Normal file
227
pkg/provider/kubernetes/gateway/grpcroute_test.go
Normal file
|
@ -0,0 +1,227 @@
|
|||
package gateway
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"k8s.io/utils/ptr"
|
||||
gatev1 "sigs.k8s.io/gateway-api/apis/v1"
|
||||
)
|
||||
|
||||
func Test_buildGRPCMatchRule(t *testing.T) {
|
||||
testCases := []struct {
|
||||
desc string
|
||||
match gatev1.GRPCRouteMatch
|
||||
hostnames []gatev1.Hostname
|
||||
expectedRule string
|
||||
expectedPriority int
|
||||
expectedError bool
|
||||
}{
|
||||
{
|
||||
desc: "Empty rule and matches",
|
||||
expectedRule: "PathPrefix(`/`)",
|
||||
expectedPriority: 15,
|
||||
},
|
||||
{
|
||||
desc: "One Host rule without match",
|
||||
hostnames: []gatev1.Hostname{"foo.com"},
|
||||
expectedRule: "Host(`foo.com`) && PathPrefix(`/`)",
|
||||
expectedPriority: 22,
|
||||
},
|
||||
{
|
||||
desc: "One GRPCRouteMatch with no GRPCHeaderMatch",
|
||||
match: gatev1.GRPCRouteMatch{
|
||||
Method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchExact),
|
||||
Service: ptr.To("foo"),
|
||||
Method: ptr.To("bar"),
|
||||
},
|
||||
},
|
||||
expectedRule: "PathRegexp(`/foo/bar`)",
|
||||
expectedPriority: 22,
|
||||
},
|
||||
{
|
||||
desc: "One GRPCRouteMatch with one GRPCHeaderMatch",
|
||||
match: gatev1.GRPCRouteMatch{
|
||||
Method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchExact),
|
||||
Service: ptr.To("foo"),
|
||||
Method: ptr.To("bar"),
|
||||
},
|
||||
Headers: []gatev1.GRPCHeaderMatch{
|
||||
{
|
||||
Type: ptr.To(gatev1.HeaderMatchExact),
|
||||
Name: "foo",
|
||||
Value: "bar",
|
||||
},
|
||||
},
|
||||
},
|
||||
expectedRule: "PathRegexp(`/foo/bar`) && Header(`foo`,`bar`)",
|
||||
expectedPriority: 45,
|
||||
},
|
||||
{
|
||||
desc: "One GRPCRouteMatch with one GRPCHeaderMatch and one Host",
|
||||
hostnames: []gatev1.Hostname{"foo.com"},
|
||||
match: gatev1.GRPCRouteMatch{
|
||||
Method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchExact),
|
||||
Service: ptr.To("foo"),
|
||||
Method: ptr.To("bar"),
|
||||
},
|
||||
Headers: []gatev1.GRPCHeaderMatch{
|
||||
{
|
||||
Type: ptr.To(gatev1.HeaderMatchExact),
|
||||
Name: "foo",
|
||||
Value: "bar",
|
||||
},
|
||||
},
|
||||
},
|
||||
expectedRule: "Host(`foo.com`) && PathRegexp(`/foo/bar`) && Header(`foo`,`bar`)",
|
||||
expectedPriority: 52,
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range testCases {
|
||||
t.Run(test.desc, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rule, priority := buildGRPCMatchRule(test.hostnames, test.match)
|
||||
assert.Equal(t, test.expectedRule, rule)
|
||||
assert.Equal(t, test.expectedPriority, priority)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_buildGRPCMethodRule(t *testing.T) {
|
||||
testCases := []struct {
|
||||
desc string
|
||||
method *gatev1.GRPCMethodMatch
|
||||
expectedRule string
|
||||
}{
|
||||
{
|
||||
desc: "Empty",
|
||||
expectedRule: "PathPrefix(`/`)",
|
||||
},
|
||||
{
|
||||
desc: "Exact service matching",
|
||||
method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchExact),
|
||||
Service: ptr.To("foo"),
|
||||
},
|
||||
expectedRule: "PathRegexp(`/foo/[^/]+`)",
|
||||
},
|
||||
{
|
||||
desc: "Exact method matching",
|
||||
method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchExact),
|
||||
Method: ptr.To("bar"),
|
||||
},
|
||||
expectedRule: "PathRegexp(`/[^/]+/bar`)",
|
||||
},
|
||||
{
|
||||
desc: "Exact service and method matching",
|
||||
method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchExact),
|
||||
Service: ptr.To("foo"),
|
||||
Method: ptr.To("bar"),
|
||||
},
|
||||
expectedRule: "PathRegexp(`/foo/bar`)",
|
||||
},
|
||||
{
|
||||
desc: "Regexp service matching",
|
||||
method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchRegularExpression),
|
||||
Service: ptr.To("[^1-9/]"),
|
||||
},
|
||||
expectedRule: "PathRegexp(`/[^1-9/]/[^/]+`)",
|
||||
},
|
||||
{
|
||||
desc: "Regexp method matching",
|
||||
method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchRegularExpression),
|
||||
Method: ptr.To("[^1-9/]"),
|
||||
},
|
||||
expectedRule: "PathRegexp(`/[^/]+/[^1-9/]`)",
|
||||
},
|
||||
{
|
||||
desc: "Regexp service and method matching",
|
||||
method: &gatev1.GRPCMethodMatch{
|
||||
Type: ptr.To(gatev1.GRPCMethodMatchRegularExpression),
|
||||
Service: ptr.To("[^1-9/]"),
|
||||
Method: ptr.To("[^1-9/]"),
|
||||
},
|
||||
expectedRule: "PathRegexp(`/[^1-9/]/[^1-9/]`)",
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range testCases {
|
||||
t.Run(test.desc, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rule := buildGRPCMethodRule(test.method)
|
||||
assert.Equal(t, test.expectedRule, rule)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_buildGRPCHeaderRules(t *testing.T) {
|
||||
testCases := []struct {
|
||||
desc string
|
||||
headers []gatev1.GRPCHeaderMatch
|
||||
expectedRules []string
|
||||
}{
|
||||
{
|
||||
desc: "Empty",
|
||||
},
|
||||
{
|
||||
desc: "One exact match type",
|
||||
headers: []gatev1.GRPCHeaderMatch{
|
||||
{
|
||||
Type: ptr.To(gatev1.HeaderMatchExact),
|
||||
Name: "foo",
|
||||
Value: "bar",
|
||||
},
|
||||
},
|
||||
expectedRules: []string{"Header(`foo`,`bar`)"},
|
||||
},
|
||||
{
|
||||
desc: "One regexp match type",
|
||||
headers: []gatev1.GRPCHeaderMatch{
|
||||
{
|
||||
Type: ptr.To(gatev1.HeaderMatchRegularExpression),
|
||||
Name: "foo",
|
||||
Value: ".*",
|
||||
},
|
||||
},
|
||||
expectedRules: []string{"HeaderRegexp(`foo`,`.*`)"},
|
||||
},
|
||||
{
|
||||
desc: "One exact and regexp match type",
|
||||
headers: []gatev1.GRPCHeaderMatch{
|
||||
{
|
||||
Type: ptr.To(gatev1.HeaderMatchExact),
|
||||
Name: "foo",
|
||||
Value: "bar",
|
||||
},
|
||||
{
|
||||
Type: ptr.To(gatev1.HeaderMatchRegularExpression),
|
||||
Name: "foo",
|
||||
Value: ".*",
|
||||
},
|
||||
},
|
||||
expectedRules: []string{
|
||||
"Header(`foo`,`bar`)",
|
||||
"HeaderRegexp(`foo`,`.*`)",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range testCases {
|
||||
t.Run(test.desc, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rule := buildGRPCHeaderRules(test.headers)
|
||||
assert.Equal(t, test.expectedRules, rule)
|
||||
})
|
||||
}
|
||||
}
|
|
@ -116,16 +116,6 @@ func (p *Provider) loadHTTPRoute(ctx context.Context, listener gatewayListener,
|
|||
Reason: string(gatev1.RouteConditionResolvedRefs),
|
||||
}
|
||||
|
||||
errWrr := dynamic.WeightedRoundRobin{
|
||||
Services: []dynamic.WRRService{
|
||||
{
|
||||
Name: "invalid-httproute-filter",
|
||||
Status: ptr.To(500),
|
||||
Weight: ptr.To(1),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for ri, routeRule := range route.Spec.Rules {
|
||||
// Adding the gateway desc and the entryPoint desc prevents overlapping of routers build from the same routes.
|
||||
routeKey := provider.Normalize(fmt.Sprintf("%s-%s-%s-%s-%d", route.Namespace, route.Name, listener.GWName, listener.EPName, ri))
|
||||
|
@ -150,7 +140,17 @@ func (p *Provider) loadHTTPRoute(ctx context.Context, listener gatewayListener,
|
|||
log.Ctx(ctx).Error().Err(err).Msg("Unable to load HTTPRoute filters")
|
||||
|
||||
errWrrName := routerName + "-err-wrr"
|
||||
conf.HTTP.Services[errWrrName] = &dynamic.Service{Weighted: &errWrr}
|
||||
conf.HTTP.Services[errWrrName] = &dynamic.Service{
|
||||
Weighted: &dynamic.WeightedRoundRobin{
|
||||
Services: []dynamic.WRRService{
|
||||
{
|
||||
Name: "invalid-httproute-filter",
|
||||
Status: ptr.To(500),
|
||||
Weight: ptr.To(1),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
router.Service = errWrrName
|
||||
|
||||
case len(routeRule.BackendRefs) == 1 && isInternalService(routeRule.BackendRefs[0].BackendRef):
|
||||
|
|
|
@ -72,7 +72,7 @@ func Test_buildHostRule(t *testing.T) {
|
|||
func Test_buildMatchRule(t *testing.T) {
|
||||
testCases := []struct {
|
||||
desc string
|
||||
routeMatch gatev1.HTTPRouteMatch
|
||||
match gatev1.HTTPRouteMatch
|
||||
hostnames []gatev1.Hostname
|
||||
expectedRule string
|
||||
expectedPriority int
|
||||
|
@ -91,7 +91,7 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
},
|
||||
{
|
||||
desc: "One HTTPRouteMatch with nil HTTPHeaderMatch",
|
||||
routeMatch: gatev1.HTTPRouteMatch{
|
||||
match: gatev1.HTTPRouteMatch{
|
||||
Path: ptr.To(gatev1.HTTPPathMatch{
|
||||
Type: ptr.To(gatev1.PathMatchPathPrefix),
|
||||
Value: ptr.To("/"),
|
||||
|
@ -103,7 +103,7 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
},
|
||||
{
|
||||
desc: "One HTTPRouteMatch with nil HTTPHeaderMatch Type",
|
||||
routeMatch: gatev1.HTTPRouteMatch{
|
||||
match: gatev1.HTTPRouteMatch{
|
||||
Path: ptr.To(gatev1.HTTPPathMatch{
|
||||
Type: ptr.To(gatev1.PathMatchPathPrefix),
|
||||
Value: ptr.To("/"),
|
||||
|
@ -117,13 +117,13 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
},
|
||||
{
|
||||
desc: "One HTTPRouteMatch with nil HTTPPathMatch",
|
||||
routeMatch: gatev1.HTTPRouteMatch{Path: nil},
|
||||
match: gatev1.HTTPRouteMatch{Path: nil},
|
||||
expectedRule: "PathPrefix(`/`)",
|
||||
expectedPriority: 1,
|
||||
},
|
||||
{
|
||||
desc: "One HTTPRouteMatch with nil HTTPPathMatch Type",
|
||||
routeMatch: gatev1.HTTPRouteMatch{
|
||||
match: gatev1.HTTPRouteMatch{
|
||||
Path: &gatev1.HTTPPathMatch{
|
||||
Type: nil,
|
||||
Value: ptr.To("/foo/"),
|
||||
|
@ -134,7 +134,7 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
},
|
||||
{
|
||||
desc: "One HTTPRouteMatch with nil HTTPPathMatch Values",
|
||||
routeMatch: gatev1.HTTPRouteMatch{
|
||||
match: gatev1.HTTPRouteMatch{
|
||||
Path: &gatev1.HTTPPathMatch{
|
||||
Type: ptr.To(gatev1.PathMatchExact),
|
||||
Value: nil,
|
||||
|
@ -145,7 +145,7 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
},
|
||||
{
|
||||
desc: "One Path",
|
||||
routeMatch: gatev1.HTTPRouteMatch{
|
||||
match: gatev1.HTTPRouteMatch{
|
||||
Path: &gatev1.HTTPPathMatch{
|
||||
Type: ptr.To(gatev1.PathMatchExact),
|
||||
Value: ptr.To("/foo/"),
|
||||
|
@ -156,7 +156,7 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
},
|
||||
{
|
||||
desc: "Path && Header",
|
||||
routeMatch: gatev1.HTTPRouteMatch{
|
||||
match: gatev1.HTTPRouteMatch{
|
||||
Path: &gatev1.HTTPPathMatch{
|
||||
Type: ptr.To(gatev1.PathMatchExact),
|
||||
Value: ptr.To("/foo/"),
|
||||
|
@ -175,7 +175,7 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
{
|
||||
desc: "Host && Path && Header",
|
||||
hostnames: []gatev1.Hostname{"foo.com"},
|
||||
routeMatch: gatev1.HTTPRouteMatch{
|
||||
match: gatev1.HTTPRouteMatch{
|
||||
Path: &gatev1.HTTPPathMatch{
|
||||
Type: ptr.To(gatev1.PathMatchExact),
|
||||
Value: ptr.To("/foo/"),
|
||||
|
@ -197,7 +197,7 @@ func Test_buildMatchRule(t *testing.T) {
|
|||
t.Run(test.desc, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rule, priority := buildMatchRule(test.hostnames, test.routeMatch)
|
||||
rule, priority := buildMatchRule(test.hostnames, test.match)
|
||||
assert.Equal(t, test.expectedRule, rule)
|
||||
assert.Equal(t, test.expectedPriority, priority)
|
||||
})
|
||||
|
|
|
@ -44,6 +44,7 @@ const (
|
|||
kindGateway = "Gateway"
|
||||
kindTraefikService = "TraefikService"
|
||||
kindHTTPRoute = "HTTPRoute"
|
||||
kindGRPCRoute = "GRPCRoute"
|
||||
kindTCPRoute = "TCPRoute"
|
||||
kindTLSRoute = "TLSRoute"
|
||||
)
|
||||
|
@ -155,8 +156,7 @@ func (p *Provider) applyRouterTransform(ctx context.Context, rt *dynamic.Router,
|
|||
return
|
||||
}
|
||||
|
||||
err := p.routerTransform.Apply(ctx, rt, route)
|
||||
if err != nil {
|
||||
if err := p.routerTransform.Apply(ctx, rt, route); err != nil {
|
||||
log.Ctx(ctx).Error().Err(err).Msg("Apply router transform")
|
||||
}
|
||||
}
|
||||
|
@ -356,6 +356,8 @@ func (p *Provider) loadConfigurationFromGateways(ctx context.Context) *dynamic.C
|
|||
|
||||
p.loadHTTPRoutes(ctx, gatewayListeners, conf)
|
||||
|
||||
p.loadGRPCRoutes(ctx, gatewayListeners, conf)
|
||||
|
||||
if p.ExperimentalChannel {
|
||||
p.loadTCPRoutes(ctx, gatewayListeners, conf)
|
||||
p.loadTLSRoutes(ctx, gatewayListeners, conf)
|
||||
|
@ -864,7 +866,10 @@ func supportedRouteKinds(protocol gatev1.ProtocolType, experimentalChannel bool)
|
|||
}}
|
||||
|
||||
case gatev1.HTTPProtocolType, gatev1.HTTPSProtocolType:
|
||||
return []gatev1.RouteGroupKind{{Kind: kindHTTPRoute, Group: &group}}, nil
|
||||
return []gatev1.RouteGroupKind{
|
||||
{Kind: kindHTTPRoute, Group: &group},
|
||||
{Kind: kindGRPCRoute, Group: &group},
|
||||
}, nil
|
||||
|
||||
case gatev1.TLSProtocolType:
|
||||
if experimentalChannel {
|
||||
|
|
|
@ -3,6 +3,7 @@ package service
|
|||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
|
@ -30,6 +31,7 @@ import (
|
|||
"github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/failover"
|
||||
"github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/mirror"
|
||||
"github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/wrr"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
const defaultMaxBodySize int64 = -1
|
||||
|
@ -222,15 +224,7 @@ func (m *Manager) getWRRServiceHandler(ctx context.Context, serviceName string,
|
|||
|
||||
balancer := wrr.New(config.Sticky, config.HealthCheck != nil)
|
||||
for _, service := range shuffle(config.Services, m.rand) {
|
||||
if service.Status != nil {
|
||||
serviceHandler := http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
||||
writer.WriteHeader(*service.Status)
|
||||
})
|
||||
balancer.Add(service.Name, serviceHandler, service.Weight)
|
||||
continue
|
||||
}
|
||||
|
||||
serviceHandler, err := m.BuildHTTP(ctx, service.Name)
|
||||
serviceHandler, err := m.getServiceHandler(ctx, service)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -260,6 +254,34 @@ func (m *Manager) getWRRServiceHandler(ctx context.Context, serviceName string,
|
|||
return balancer, nil
|
||||
}
|
||||
|
||||
func (m *Manager) getServiceHandler(ctx context.Context, service dynamic.WRRService) (http.Handler, error) {
|
||||
switch {
|
||||
case service.Status != nil:
|
||||
return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
||||
rw.WriteHeader(*service.Status)
|
||||
}), nil
|
||||
|
||||
case service.GRPCStatus != nil:
|
||||
return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
||||
st := status.New(service.GRPCStatus.Code, service.GRPCStatus.Msg)
|
||||
|
||||
body, err := json.Marshal(st.Proto())
|
||||
if err != nil {
|
||||
http.Error(rw, "Failed to marshal status to JSON", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
rw.Header().Set("Content-Type", "application/json")
|
||||
rw.WriteHeader(http.StatusOK)
|
||||
|
||||
_, _ = rw.Write(body)
|
||||
}), nil
|
||||
|
||||
default:
|
||||
return m.BuildHTTP(ctx, service.Name)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) getLoadBalancerServiceHandler(ctx context.Context, serviceName string, info *runtime.ServiceInfo) (http.Handler, error) {
|
||||
service := info.LoadBalancer
|
||||
|
||||
|
|
Loading…
Reference in a new issue