This is an automated email from the ASF dual-hosted git repository.
alinsran 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 95787e6e chore: refactor provider (#2507)
95787e6e is described below
commit 95787e6e68309f136cda80b546e4cbe8b5bccffc
Author: AlinsRan <alins...@apache.org>
AuthorDate: Tue Aug 19 08:29:57 2025 +0800
chore: refactor provider (#2507)
---
.github/workflows/apisix-conformance-test.yml | 2 +-
api/adc/types.go | 21 +
internal/{provider => }/adc/cache/cache.go | 0
internal/{provider => }/adc/cache/indexer.go | 0
internal/{provider => }/adc/cache/memdb.go | 0
internal/{provider => }/adc/cache/noop_db.go | 0
internal/{provider => }/adc/cache/schema.go | 0
internal/{provider/adc => adc/cache}/store.go | 15 +-
internal/adc/client/client.go | 317 +++++++++++++
internal/{provider/adc => adc/client}/executor.go | 12 +-
internal/{provider => }/adc/options.go | 0
.../adc/translator/apisixconsumer.go | 0
.../{provider => }/adc/translator/apisixroute.go | 0
.../{provider => }/adc/translator/apisixtls.go | 0
.../adc/translator/apisixupstream.go | 0
internal/{provider => }/adc/translator/consumer.go | 0
internal/{provider => }/adc/translator/gateway.go | 0
.../config.go => adc/translator/gatewayproxy.go} | 124 +-----
.../{provider => }/adc/translator/globalrule.go | 0
.../{provider => }/adc/translator/httproute.go | 0
internal/{provider => }/adc/translator/ingress.go | 0
.../{provider => }/adc/translator/ingressclass.go | 0
internal/{provider => }/adc/translator/policies.go | 0
.../{provider => }/adc/translator/translator.go | 7 +
internal/manager/run.go | 7 +-
internal/provider/adc/adc.go | 490 ---------------------
internal/provider/apisix/provider.go | 297 +++++++++++++
internal/provider/{adc => apisix}/status.go | 23 +-
internal/provider/common/configmanager.go | 170 +++++++
.../{adc/translator/translator.go => init/init.go} | 26 +-
internal/provider/{adc => }/options.go | 34 +-
internal/provider/provider.go | 1 -
internal/provider/register.go | 49 +++
internal/types/types.go | 29 +-
test/e2e/crds/v1alpha1/gatewayproxy.go | 3 +-
test/e2e/framework/apisix_consts.go | 5 +
.../e2e/framework/manifests/apisix-standalone.yaml | 4 +-
test/e2e/framework/manifests/apisix.yaml | 4 +-
test/e2e/scaffold/adc.go | 2 +-
test/e2e/scaffold/apisix_deployer.go | 3 +-
40 files changed, 991 insertions(+), 654 deletions(-)
diff --git a/.github/workflows/apisix-conformance-test.yml
b/.github/workflows/apisix-conformance-test.yml
index 8549fb19..d800252c 100644
--- a/.github/workflows/apisix-conformance-test.yml
+++ b/.github/workflows/apisix-conformance-test.yml
@@ -89,7 +89,7 @@ jobs:
- name: Install And Run Cloud Provider KIND
run: |
- go install sigs.k8s.io/cloud-provider-kind@latest
+ go install sigs.k8s.io/cloud-provider-kind@v0.6.0
nohup cloud-provider-kind > /tmp/kind-loadbalancer.log 2>&1 &
- name: Install Gateway API And CRDs
diff --git a/api/adc/types.go b/api/adc/types.go
index d561abf8..2f6a020d 100644
--- a/api/adc/types.go
+++ b/api/adc/types.go
@@ -726,3 +726,24 @@ func (s *StringOrSlice) UnmarshalJSON(p []byte) error {
}
return json.Unmarshal(p, &s.StrVal)
}
+
+type Config struct {
+ Name string
+ ServerAddrs []string
+ Token string
+ TlsVerify bool
+}
+
+// MarshalJSON implements custom JSON marshaling for adcConfig
+// It excludes the Token field for security reasons
+func (c Config) MarshalJSON() ([]byte, error) {
+ return json.Marshal(struct {
+ Name string `json:"name"`
+ ServerAddrs []string `json:"serverAddrs"`
+ TlsVerify bool `json:"tlsVerify"`
+ }{
+ Name: c.Name,
+ ServerAddrs: c.ServerAddrs,
+ TlsVerify: c.TlsVerify,
+ })
+}
diff --git a/internal/provider/adc/cache/cache.go b/internal/adc/cache/cache.go
similarity index 100%
rename from internal/provider/adc/cache/cache.go
rename to internal/adc/cache/cache.go
diff --git a/internal/provider/adc/cache/indexer.go
b/internal/adc/cache/indexer.go
similarity index 100%
rename from internal/provider/adc/cache/indexer.go
rename to internal/adc/cache/indexer.go
diff --git a/internal/provider/adc/cache/memdb.go b/internal/adc/cache/memdb.go
similarity index 100%
rename from internal/provider/adc/cache/memdb.go
rename to internal/adc/cache/memdb.go
diff --git a/internal/provider/adc/cache/noop_db.go
b/internal/adc/cache/noop_db.go
similarity index 100%
rename from internal/provider/adc/cache/noop_db.go
rename to internal/adc/cache/noop_db.go
diff --git a/internal/provider/adc/cache/schema.go
b/internal/adc/cache/schema.go
similarity index 100%
rename from internal/provider/adc/cache/schema.go
rename to internal/adc/cache/schema.go
diff --git a/internal/provider/adc/store.go b/internal/adc/cache/store.go
similarity index 96%
rename from internal/provider/adc/store.go
rename to internal/adc/cache/store.go
index 458a68a2..70152266 100644
--- a/internal/provider/adc/store.go
+++ b/internal/adc/cache/store.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package adc
+package cache
import (
"fmt"
@@ -27,11 +27,10 @@ import (
adctypes "github.com/apache/apisix-ingress-controller/api/adc"
"github.com/apache/apisix-ingress-controller/internal/controller/label"
-
"github.com/apache/apisix-ingress-controller/internal/provider/adc/cache"
)
type Store struct {
- cacheMap map[string]cache.Cache
+ cacheMap map[string]Cache
pluginMetadataMap map[string]adctypes.PluginMetadata
sync.Mutex
@@ -39,17 +38,17 @@ type Store struct {
func NewStore() *Store {
return &Store{
- cacheMap: make(map[string]cache.Cache),
+ cacheMap: make(map[string]Cache),
pluginMetadataMap: make(map[string]adctypes.PluginMetadata),
}
}
-func (s *Store) Insert(name string, resourceTypes []string, resources
adctypes.Resources, Labels map[string]string) error {
+func (s *Store) Insert(name string, resourceTypes []string, resources
*adctypes.Resources, Labels map[string]string) error {
s.Lock()
defer s.Unlock()
targetCache, ok := s.cacheMap[name]
if !ok {
- db, err := cache.NewMemDBCache()
+ db, err := NewMemDBCache()
if err != nil {
return err
}
@@ -57,7 +56,7 @@ func (s *Store) Insert(name string, resourceTypes []string,
resources adctypes.R
targetCache = s.cacheMap[name]
}
log.Debugw("Inserting resources into cache for", zap.String("name",
name))
- selector := &cache.KindLabelSelector{
+ selector := &KindLabelSelector{
Kind: Labels[label.LabelKind],
Name: Labels[label.LabelName],
Namespace: Labels[label.LabelNamespace],
@@ -153,7 +152,7 @@ func (s *Store) Delete(name string, resourceTypes []string,
Labels map[string]st
if !ok {
return nil
}
- selector := &cache.KindLabelSelector{
+ selector := &KindLabelSelector{
Kind: Labels[label.LabelKind],
Name: Labels[label.LabelName],
Namespace: Labels[label.LabelNamespace],
diff --git a/internal/adc/client/client.go b/internal/adc/client/client.go
new file mode 100644
index 00000000..93beb47f
--- /dev/null
+++ b/internal/adc/client/client.go
@@ -0,0 +1,317 @@
+// 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 client
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "os"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/api7/gopkg/pkg/log"
+ "github.com/pkg/errors"
+ "go.uber.org/zap"
+
+ adctypes "github.com/apache/apisix-ingress-controller/api/adc"
+ "github.com/apache/apisix-ingress-controller/internal/adc/cache"
+ "github.com/apache/apisix-ingress-controller/internal/provider/common"
+ "github.com/apache/apisix-ingress-controller/internal/types"
+ pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
+)
+
+type Client struct {
+ syncMu sync.RWMutex
+ mu sync.Mutex
+ *cache.Store
+
+ executor ADCExecutor
+ BackendMode string
+
+ ConfigManager *common.ConfigManager[types.NamespacedNameKind,
adctypes.Config]
+}
+
+func New(mode string, timeout time.Duration) (*Client, error) {
+ return &Client{
+ Store: cache.NewStore(),
+ executor: &DefaultADCExecutor{},
+ BackendMode: mode,
+ ConfigManager:
common.NewConfigManager[types.NamespacedNameKind, adctypes.Config](),
+ }, nil
+}
+
+type Task struct {
+ Key types.NamespacedNameKind
+ Name string
+ Labels map[string]string
+ Configs map[types.NamespacedNameKind]adctypes.Config
+ ResourceTypes []string
+ Resources *adctypes.Resources
+}
+
+func (d *Client) Update(ctx context.Context, args Task) error {
+ d.mu.Lock()
+ deleteConfigs := d.ConfigManager.Update(args.Key, args.Configs)
+ for _, config := range deleteConfigs {
+ if err := d.Store.Delete(config.Name, args.ResourceTypes,
args.Labels); err != nil {
+ log.Errorw("failed to delete resources from store",
+ zap.String("name", config.Name),
+ zap.Error(err),
+ )
+ return err
+ }
+ }
+
+ for _, config := range args.Configs {
+ if err := d.Insert(config.Name, args.ResourceTypes,
args.Resources, args.Labels); err != nil {
+ log.Errorw("failed to insert resources into store",
+ zap.String("name", config.Name),
+ zap.Error(err),
+ )
+ return err
+ }
+ }
+ d.mu.Unlock()
+
+ d.syncMu.RLock()
+ defer d.syncMu.RUnlock()
+
+ if len(deleteConfigs) > 0 {
+ err := d.sync(ctx, Task{
+ Name: args.Name,
+ Labels: args.Labels,
+ ResourceTypes: args.ResourceTypes,
+ Configs: deleteConfigs,
+ })
+ if err != nil {
+ log.Warnw("failed to sync deleted configs",
zap.Error(err))
+ }
+ }
+
+ return d.sync(ctx, args)
+}
+
+func (d *Client) UpdateConfig(ctx context.Context, args Task) error {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ deleteConfigs := d.ConfigManager.Update(args.Key, args.Configs)
+
+ for _, config := range deleteConfigs {
+ if err := d.Store.Delete(config.Name, args.ResourceTypes,
args.Labels); err != nil {
+ log.Errorw("failed to delete resources from store",
+ zap.String("name", config.Name),
+ zap.Error(err),
+ )
+ return err
+ }
+ }
+
+ for _, config := range args.Configs {
+ if err := d.Insert(config.Name, args.ResourceTypes,
args.Resources, args.Labels); err != nil {
+ log.Errorw("failed to insert resources into store",
+ zap.String("name", config.Name),
+ zap.Error(err),
+ )
+ return err
+ }
+ }
+ return nil
+}
+
+func (d *Client) Delete(ctx context.Context, args Task) error {
+ d.mu.Lock()
+ configs := d.ConfigManager.Get(args.Key)
+ d.ConfigManager.Delete(args.Key)
+
+ for _, config := range configs {
+ if err := d.Store.Delete(config.Name, args.ResourceTypes,
args.Labels); err != nil {
+ log.Errorw("failed to delete resources from store",
+ zap.String("name", config.Name),
+ zap.Error(err),
+ )
+ return err
+ }
+ }
+ d.mu.Unlock()
+
+ d.syncMu.RLock()
+ defer d.syncMu.RUnlock()
+
+ return d.sync(ctx, Task{
+ Labels: args.Labels,
+ ResourceTypes: args.ResourceTypes,
+ Configs: configs,
+ })
+}
+
+func (d *Client) DeleteConfig(ctx context.Context, args Task) error {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+
+ configs := d.ConfigManager.Get(args.Key)
+ d.ConfigManager.Delete(args.Key)
+
+ for _, config := range configs {
+ if err := d.Store.Delete(config.Name, args.ResourceTypes,
args.Labels); err != nil {
+ log.Errorw("failed to delete resources from store",
+ zap.String("name", config.Name),
+ zap.Error(err),
+ )
+ return err
+ }
+ }
+
+ return nil
+}
+
+func (c *Client) Sync(ctx context.Context)
(map[string]types.ADCExecutionErrors, error) {
+ c.syncMu.Lock()
+ defer c.syncMu.Unlock()
+ log.Debug("syncing all resources")
+
+ configs := c.ConfigManager.List()
+
+ if len(configs) == 0 {
+ log.Warn("no GatewayProxy configs provided")
+ return nil, nil
+ }
+
+ log.Debugw("syncing resources with multiple configs",
zap.Any("configs", configs))
+
+ failedMap := map[string]types.ADCExecutionErrors{}
+ var failedConfigs []string
+ for _, config := range configs {
+ name := config.Name
+ resources, err := c.GetResources(name)
+ if err != nil {
+ log.Errorw("failed to get resources from store",
zap.String("name", name), zap.Error(err))
+ failedConfigs = append(failedConfigs, name)
+ continue
+ }
+ if resources == nil {
+ continue
+ }
+
+ if err := c.sync(ctx, Task{
+ Name: name + "-sync",
+ Configs: map[types.NamespacedNameKind]adctypes.Config{
+ {}: config,
+ },
+ Resources: resources,
+ }); err != nil {
+ log.Errorw("failed to sync resources",
zap.String("name", name), zap.Error(err))
+ failedConfigs = append(failedConfigs, name)
+ var execErrs types.ADCExecutionErrors
+ if errors.As(err, &execErrs) {
+ failedMap[name] = execErrs
+ }
+ }
+ }
+
+ var err error
+ if len(failedConfigs) > 0 {
+ err = fmt.Errorf("failed to sync %d configs: %s",
+ len(failedConfigs),
+ strings.Join(failedConfigs, ", "))
+ }
+ return failedMap, err
+}
+
+func (c *Client) sync(ctx context.Context, task Task) error {
+ log.Debugw("syncing resources", zap.Any("task", task))
+
+ if len(task.Configs) == 0 {
+ log.Warnw("no adc configs provided", zap.Any("task", task))
+ return nil
+ }
+
+ var errs types.ADCExecutionErrors
+
+ // Record file I/O duration
+ fileIOStart := time.Now()
+ // every task resources is the same, so we can use the first config to
prepare the sync file
+ syncFilePath, cleanup, err := prepareSyncFile(task.Resources)
+ if err != nil {
+ pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure",
time.Since(fileIOStart).Seconds())
+ return err
+ }
+ pkgmetrics.RecordFileIODuration("prepare_sync_file", "success",
time.Since(fileIOStart).Seconds())
+ defer cleanup()
+
+ args := BuildADCExecuteArgs(syncFilePath, task.Labels,
task.ResourceTypes)
+
+ for _, config := range task.Configs {
+ // Record sync duration for each config
+ startTime := time.Now()
+ resourceType := strings.Join(task.ResourceTypes, ",")
+ if resourceType == "" {
+ resourceType = "all"
+ }
+
+ err := c.executor.Execute(ctx, c.BackendMode, config, args)
+ duration := time.Since(startTime).Seconds()
+
+ status := "success"
+ if err != nil {
+ status = "failure"
+ log.Errorw("failed to execute adc command",
zap.Error(err), zap.Any("config", config))
+
+ var execErr types.ADCExecutionError
+ if errors.As(err, &execErr) {
+ errs.Errors = append(errs.Errors, execErr)
+ pkgmetrics.RecordExecutionError(config.Name,
execErr.Name)
+ } else {
+ pkgmetrics.RecordExecutionError(config.Name,
"unknown")
+ }
+ }
+
+ // Record metrics
+ pkgmetrics.RecordSyncDuration(config.Name, resourceType,
status, duration)
+ }
+
+ if len(errs.Errors) > 0 {
+ return errs
+ }
+ return nil
+}
+
+func prepareSyncFile(resources any) (string, func(), error) {
+ data, err := json.Marshal(resources)
+ if err != nil {
+ return "", nil, err
+ }
+
+ tmpFile, err := os.CreateTemp("", "adc-task-*.json")
+ if err != nil {
+ return "", nil, err
+ }
+ cleanup := func() {
+ _ = tmpFile.Close()
+ _ = os.Remove(tmpFile.Name())
+ }
+ if _, err := tmpFile.Write(data); err != nil {
+ cleanup()
+ return "", nil, err
+ }
+
+ log.Debugw("generated adc file", zap.String("filename",
tmpFile.Name()), zap.String("json", string(data)))
+
+ return tmpFile.Name(), cleanup, nil
+}
diff --git a/internal/provider/adc/executor.go b/internal/adc/client/executor.go
similarity index 93%
rename from internal/provider/adc/executor.go
rename to internal/adc/client/executor.go
index 5377018d..56b1b02c 100644
--- a/internal/provider/adc/executor.go
+++ b/internal/adc/client/executor.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package adc
+package client
import (
"bytes"
@@ -37,21 +37,21 @@ import (
)
type ADCExecutor interface {
- Execute(ctx context.Context, mode string, config adcConfig, args
[]string) error
+ Execute(ctx context.Context, mode string, config adctypes.Config, args
[]string) error
}
type DefaultADCExecutor struct {
sync.Mutex
}
-func (e *DefaultADCExecutor) Execute(ctx context.Context, mode string, config
adcConfig, args []string) error {
+func (e *DefaultADCExecutor) Execute(ctx context.Context, mode string, config
adctypes.Config, args []string) error {
e.Lock()
defer e.Unlock()
return e.runADC(ctx, mode, config, args)
}
-func (e *DefaultADCExecutor) runADC(ctx context.Context, mode string, config
adcConfig, args []string) error {
+func (e *DefaultADCExecutor) runADC(ctx context.Context, mode string, config
adctypes.Config, args []string) error {
var execErrs = types.ADCExecutionError{
Name: config.Name,
}
@@ -76,13 +76,13 @@ func (e *DefaultADCExecutor) runADC(ctx context.Context,
mode string, config adc
return nil
}
-func (e *DefaultADCExecutor) runForSingleServerWithTimeout(ctx
context.Context, serverAddr, mode string, config adcConfig, args []string)
error {
+func (e *DefaultADCExecutor) runForSingleServerWithTimeout(ctx
context.Context, serverAddr, mode string, config adctypes.Config, args
[]string) error {
ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
return e.runForSingleServer(ctx, serverAddr, mode, config, args)
}
-func (e *DefaultADCExecutor) runForSingleServer(ctx context.Context,
serverAddr, mode string, config adcConfig, args []string) error {
+func (e *DefaultADCExecutor) runForSingleServer(ctx context.Context,
serverAddr, mode string, config adctypes.Config, args []string) error {
cmdArgs := append([]string{}, args...)
if !config.TlsVerify {
cmdArgs = append(cmdArgs, "--tls-skip-verify")
diff --git a/internal/provider/adc/options.go b/internal/adc/options.go
similarity index 100%
copy from internal/provider/adc/options.go
copy to internal/adc/options.go
diff --git a/internal/provider/adc/translator/apisixconsumer.go
b/internal/adc/translator/apisixconsumer.go
similarity index 100%
rename from internal/provider/adc/translator/apisixconsumer.go
rename to internal/adc/translator/apisixconsumer.go
diff --git a/internal/provider/adc/translator/apisixroute.go
b/internal/adc/translator/apisixroute.go
similarity index 100%
rename from internal/provider/adc/translator/apisixroute.go
rename to internal/adc/translator/apisixroute.go
diff --git a/internal/provider/adc/translator/apisixtls.go
b/internal/adc/translator/apisixtls.go
similarity index 100%
rename from internal/provider/adc/translator/apisixtls.go
rename to internal/adc/translator/apisixtls.go
diff --git a/internal/provider/adc/translator/apisixupstream.go
b/internal/adc/translator/apisixupstream.go
similarity index 100%
rename from internal/provider/adc/translator/apisixupstream.go
rename to internal/adc/translator/apisixupstream.go
diff --git a/internal/provider/adc/translator/consumer.go
b/internal/adc/translator/consumer.go
similarity index 100%
rename from internal/provider/adc/translator/consumer.go
rename to internal/adc/translator/consumer.go
diff --git a/internal/provider/adc/translator/gateway.go
b/internal/adc/translator/gateway.go
similarity index 100%
rename from internal/provider/adc/translator/gateway.go
rename to internal/adc/translator/gateway.go
diff --git a/internal/provider/adc/config.go
b/internal/adc/translator/gatewayproxy.go
similarity index 55%
rename from internal/provider/adc/config.go
rename to internal/adc/translator/gatewayproxy.go
index 20bc3f64..8b7fb673 100644
--- a/internal/provider/adc/config.go
+++ b/internal/adc/translator/gatewayproxy.go
@@ -6,7 +6,7 @@
// "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
+// 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
@@ -15,29 +15,29 @@
// specific language governing permissions and limitations
// under the License.
-package adc
+package translator
import (
- "errors"
"fmt"
"net"
- "slices"
"strconv"
"github.com/api7/gopkg/pkg/log"
+ "github.com/pkg/errors"
"go.uber.org/zap"
+ corev1 "k8s.io/api/core/v1"
discoveryv1 "k8s.io/api/discovery/v1"
k8stypes "k8s.io/apimachinery/pkg/types"
"k8s.io/utils/ptr"
gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+ types "github.com/apache/apisix-ingress-controller/api/adc"
"github.com/apache/apisix-ingress-controller/api/v1alpha1"
"github.com/apache/apisix-ingress-controller/internal/provider"
- "github.com/apache/apisix-ingress-controller/internal/types"
"github.com/apache/apisix-ingress-controller/internal/utils"
)
-func (d *adcClient) getConfigsForGatewayProxy(tctx *provider.TranslateContext,
gatewayProxy *v1alpha1.GatewayProxy) (*adcConfig, error) {
+func (t *Translator) TranslateGatewayProxyToConfig(tctx
*provider.TranslateContext, gatewayProxy *v1alpha1.GatewayProxy,
resolveEndpoints bool) (*types.Config, error) {
if gatewayProxy == nil || gatewayProxy.Spec.Provider == nil {
return nil, nil
}
@@ -47,8 +47,8 @@ func (d *adcClient) getConfigsForGatewayProxy(tctx
*provider.TranslateContext, g
return nil, nil
}
- config := adcConfig{
- Name: k8stypes.NamespacedName{Namespace:
gatewayProxy.Namespace, Name: gatewayProxy.Name}.String(),
+ config := types.Config{
+ Name: utils.NamespacedNameKind(gatewayProxy).String(),
}
if provider.ControlPlane.TlsVerify != nil {
@@ -88,19 +88,19 @@ func (d *adcClient) getConfigsForGatewayProxy(tctx
*provider.TranslateContext, g
Namespace: gatewayProxy.Namespace,
Name: provider.ControlPlane.Service.Name,
}
- _, ok := tctx.Services[namespacedName]
+ svc, ok := tctx.Services[namespacedName]
if !ok {
return nil, fmt.Errorf("no service found for service
reference: %s", namespacedName)
}
// APISIXStandalone, configurations need to be sent to each
data plane instance;
// In other cases, the service is directly accessed as the adc
backend server address.
- if d.BackendMode == BackendModeAPISIXStandalone {
+ if resolveEndpoints {
endpoint := tctx.EndpointSlices[namespacedName]
if endpoint == nil {
return nil, nil
}
- upstreamNodes, err :=
d.translator.TranslateBackendRefWithFilter(tctx, gatewayv1.BackendRef{
+ upstreamNodes, err :=
t.TranslateBackendRefWithFilter(tctx, gatewayv1.BackendRef{
BackendObjectReference:
gatewayv1.BackendObjectReference{
Name:
gatewayv1.ObjectName(provider.ControlPlane.Service.Name),
Namespace:
(*gatewayv1.Namespace)(&gatewayProxy.Namespace),
@@ -120,9 +120,14 @@ func (d *adcClient) getConfigsForGatewayProxy(tctx
*provider.TranslateContext, g
config.ServerAddrs = append(config.ServerAddrs,
"http://"+net.JoinHostPort(node.Host, strconv.Itoa(node.Port)))
}
} else {
- config.ServerAddrs = []string{
- fmt.Sprintf("http://%s.%s:%d";,
provider.ControlPlane.Service.Name, gatewayProxy.Namespace,
provider.ControlPlane.Service.Port),
+ refPort := provider.ControlPlane.Service.Port
+ var serverAddr string
+ if svc.Spec.Type == corev1.ServiceTypeExternalName {
+ serverAddr = fmt.Sprintf("http://%s:%d";,
svc.Spec.ExternalName, refPort)
+ } else {
+ serverAddr = fmt.Sprintf("http://%s.%s.svc:%d";,
provider.ControlPlane.Service.Name, gatewayProxy.Namespace, refPort)
}
+ config.ServerAddrs = []string{serverAddr}
}
log.Debugw("add server address to config.ServiceAddrs",
zap.Strings("config.ServerAddrs", config.ServerAddrs))
@@ -130,96 +135,3 @@ func (d *adcClient) getConfigsForGatewayProxy(tctx
*provider.TranslateContext, g
return &config, nil
}
-
-func (d *adcClient) deleteConfigs(rk types.NamespacedNameKind) {
- d.Lock()
- defer d.Unlock()
- delete(d.configs, rk)
- delete(d.parentRefs, rk)
-}
-
-func (d *adcClient) getParentRefs(rk types.NamespacedNameKind)
[]types.NamespacedNameKind {
- d.Lock()
- defer d.Unlock()
- return d.parentRefs[rk]
-}
-
-func (d *adcClient) getConfigs(rk types.NamespacedNameKind) []adcConfig {
- d.Lock()
- defer d.Unlock()
- parentRefs := d.parentRefs[rk]
- configs := make([]adcConfig, 0, len(parentRefs))
- for _, parentRef := range parentRefs {
- if config, ok := d.configs[parentRef]; ok {
- configs = append(configs, config)
- }
- }
- return configs
-}
-
-func (d *adcClient) updateConfigs(rk types.NamespacedNameKind, tctx
*provider.TranslateContext) error {
- d.Lock()
- defer d.Unlock()
-
- // set parent refs
- d.parentRefs[rk] = tctx.ResourceParentRefs[rk]
- parentRefs := d.parentRefs[rk]
-
- for _, parentRef := range parentRefs {
- gatewayProxy, ok := tctx.GatewayProxies[parentRef]
- if !ok {
- log.Debugw("no gateway proxy found for parent ref",
zap.Any("parentRef", parentRef))
- continue
- }
- config, err := d.getConfigsForGatewayProxy(tctx, &gatewayProxy)
- if err != nil {
- return err
- }
- if config == nil {
- log.Debugw("no config found for gateway proxy",
zap.Any("parentRef", parentRef))
- continue
- }
- d.configs[parentRef] = *config
- }
-
- return nil
-}
-
-// updateConfigForGatewayProxy update config for all referrers of the
GatewayProxy
-func (d *adcClient) updateConfigForGatewayProxy(tctx
*provider.TranslateContext, gp *v1alpha1.GatewayProxy) error {
- d.Lock()
- defer d.Unlock()
-
- config, err := d.getConfigsForGatewayProxy(tctx, gp)
- if err != nil {
- return err
- }
-
- referrers := tctx.GatewayProxyReferrers[utils.NamespacedName(gp)]
-
- if config == nil {
- for _, ref := range referrers {
- delete(d.configs, ref)
- }
- return nil
- }
-
- for _, ref := range referrers {
- d.configs[ref] = *config
- }
-
- d.syncNotify()
- return nil
-}
-
-func (d *adcClient) findConfigsToDelete(oldParentRefs, newParentRefs
[]types.NamespacedNameKind) []adcConfig {
- var deleteConfigs []adcConfig
- for _, parentRef := range oldParentRefs {
- if !slices.ContainsFunc(newParentRefs, func(rk
types.NamespacedNameKind) bool {
- return rk.Kind == parentRef.Kind && rk.Namespace ==
parentRef.Namespace && rk.Name == parentRef.Name
- }) {
- deleteConfigs = append(deleteConfigs,
d.configs[parentRef])
- }
- }
- return deleteConfigs
-}
diff --git a/internal/provider/adc/translator/globalrule.go
b/internal/adc/translator/globalrule.go
similarity index 100%
rename from internal/provider/adc/translator/globalrule.go
rename to internal/adc/translator/globalrule.go
diff --git a/internal/provider/adc/translator/httproute.go
b/internal/adc/translator/httproute.go
similarity index 100%
rename from internal/provider/adc/translator/httproute.go
rename to internal/adc/translator/httproute.go
diff --git a/internal/provider/adc/translator/ingress.go
b/internal/adc/translator/ingress.go
similarity index 100%
rename from internal/provider/adc/translator/ingress.go
rename to internal/adc/translator/ingress.go
diff --git a/internal/provider/adc/translator/ingressclass.go
b/internal/adc/translator/ingressclass.go
similarity index 100%
rename from internal/provider/adc/translator/ingressclass.go
rename to internal/adc/translator/ingressclass.go
diff --git a/internal/provider/adc/translator/policies.go
b/internal/adc/translator/policies.go
similarity index 100%
rename from internal/provider/adc/translator/policies.go
rename to internal/adc/translator/policies.go
diff --git a/internal/provider/adc/translator/translator.go
b/internal/adc/translator/translator.go
similarity index 93%
copy from internal/provider/adc/translator/translator.go
copy to internal/adc/translator/translator.go
index 67ab7596..0d50661a 100644
--- a/internal/provider/adc/translator/translator.go
+++ b/internal/adc/translator/translator.go
@@ -26,6 +26,13 @@ import (
type Translator struct {
Log logr.Logger
}
+
+func NewTranslator(log logr.Logger) *Translator {
+ return &Translator{
+ Log: log,
+ }
+}
+
type TranslateResult struct {
Routes []*adctypes.Route
Services []*adctypes.Service
diff --git a/internal/manager/run.go b/internal/manager/run.go
index b75b383d..c08ccaba 100644
--- a/internal/manager/run.go
+++ b/internal/manager/run.go
@@ -42,7 +42,8 @@ import (
"github.com/apache/apisix-ingress-controller/internal/controller/config"
"github.com/apache/apisix-ingress-controller/internal/controller/status"
"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
- "github.com/apache/apisix-ingress-controller/internal/provider/adc"
+ "github.com/apache/apisix-ingress-controller/internal/provider"
+ _ "github.com/apache/apisix-ingress-controller/internal/provider/init"
_ "github.com/apache/apisix-ingress-controller/pkg/metrics"
)
@@ -164,7 +165,9 @@ func Run(ctx context.Context, logger logr.Logger) error {
return err
}
- provider, err := adc.New(updater.Writer(), readier, &adc.Options{
+ providerType := string(config.ControllerConfig.ProviderConfig.Type)
+
+ provider, err := provider.New(providerType, updater.Writer(), readier,
&provider.Options{
SyncTimeout: config.ControllerConfig.ExecADCTimeout.Duration,
SyncPeriod:
config.ControllerConfig.ProviderConfig.SyncPeriod.Duration,
InitSyncDelay:
config.ControllerConfig.ProviderConfig.InitSyncDelay.Duration,
diff --git a/internal/provider/adc/adc.go b/internal/provider/adc/adc.go
deleted file mode 100644
index 5ff00761..00000000
--- a/internal/provider/adc/adc.go
+++ /dev/null
@@ -1,490 +0,0 @@
-// 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 adc
-
-import (
- "context"
- "encoding/json"
- "fmt"
- "os"
- "strings"
- "sync"
- "time"
-
- "github.com/api7/gopkg/pkg/log"
- "github.com/pkg/errors"
- "go.uber.org/zap"
- networkingv1 "k8s.io/api/networking/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"
- "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"
- "github.com/apache/apisix-ingress-controller/internal/manager/readiness"
- "github.com/apache/apisix-ingress-controller/internal/provider"
-
"github.com/apache/apisix-ingress-controller/internal/provider/adc/translator"
- "github.com/apache/apisix-ingress-controller/internal/types"
- "github.com/apache/apisix-ingress-controller/internal/utils"
- pkgmetrics "github.com/apache/apisix-ingress-controller/pkg/metrics"
-)
-
-type adcConfig struct {
- Name string
- ServerAddrs []string
- Token string
- TlsVerify bool
-}
-
-// MarshalJSON implements custom JSON marshaling for adcConfig
-// It excludes the Token field for security reasons
-func (c adcConfig) MarshalJSON() ([]byte, error) {
- return json.Marshal(struct {
- Name string `json:"name"`
- ServerAddrs []string `json:"serverAddrs"`
- TlsVerify bool `json:"tlsVerify"`
- }{
- Name: c.Name,
- ServerAddrs: c.ServerAddrs,
- TlsVerify: c.TlsVerify,
- })
-}
-
-type BackendMode string
-
-const (
- BackendModeAPISIXStandalone string = "apisix-standalone"
- BackendModeAPISIX string = "apisix"
-)
-
-type adcClient struct {
- sync.Mutex
-
- syncLock sync.Mutex
-
- translator *translator.Translator
- // gateway/ingressclass -> adcConfig
- configs map[types.NamespacedNameKind]adcConfig
- // httproute/consumer/ingress/gateway -> gateway/ingressclass
- parentRefs map[types.NamespacedNameKind][]types.NamespacedNameKind
-
- store *Store
-
- executor ADCExecutor
-
- Options
-
- updater status.Updater
- statusUpdateMap map[types.NamespacedNameKind][]string
-
- readier readiness.ReadinessManager
-
- syncCh chan struct{}
-}
-
-type Task struct {
- Name string
- Resources adctypes.Resources
- Labels map[string]string
- ResourceTypes []string
- configs []adcConfig
-}
-
-func New(updater status.Updater, readier readiness.ReadinessManager, opts
...Option) (provider.Provider, error) {
- o := Options{}
- o.ApplyOptions(opts)
-
- return &adcClient{
- Options: o,
- translator: &translator.Translator{},
- configs: make(map[types.NamespacedNameKind]adcConfig),
- parentRefs:
make(map[types.NamespacedNameKind][]types.NamespacedNameKind),
- store: NewStore(),
- executor: &DefaultADCExecutor{},
- updater: updater,
- readier: readier,
- syncCh: make(chan struct{}, 1),
- }, nil
-}
-
-func (d *adcClient) Update(ctx context.Context, tctx
*provider.TranslateContext, obj client.Object) error {
- log.Debugw("updating object", zap.Any("object", obj))
- var (
- result *translator.TranslateResult
- resourceTypes []string
- err error
- )
-
- rk := utils.NamespacedNameKind(obj)
-
- switch t := obj.(type) {
- case *gatewayv1.HTTPRoute:
- result, err = d.translator.TranslateHTTPRoute(tctx,
t.DeepCopy())
- resourceTypes = append(resourceTypes, "service")
- case *gatewayv1.Gateway:
- result, err = d.translator.TranslateGateway(tctx, t.DeepCopy())
- resourceTypes = append(resourceTypes, "global_rule", "ssl",
"plugin_metadata")
- case *networkingv1.Ingress:
- result, err = d.translator.TranslateIngress(tctx, t.DeepCopy())
- resourceTypes = append(resourceTypes, "service", "ssl")
- case *v1alpha1.Consumer:
- result, err = d.translator.TranslateConsumerV1alpha1(tctx,
t.DeepCopy())
- resourceTypes = append(resourceTypes, "consumer")
- case *networkingv1.IngressClass:
- result, err = d.translator.TranslateIngressClass(tctx,
t.DeepCopy())
- resourceTypes = append(resourceTypes, "global_rule",
"plugin_metadata")
- case *apiv2.ApisixRoute:
- result, err = d.translator.TranslateApisixRoute(tctx,
t.DeepCopy())
- resourceTypes = append(resourceTypes, "service")
- case *apiv2.ApisixGlobalRule:
- result, err = d.translator.TranslateApisixGlobalRule(tctx,
t.DeepCopy())
- resourceTypes = append(resourceTypes, "global_rule")
- case *apiv2.ApisixTls:
- result, err = d.translator.TranslateApisixTls(tctx,
t.DeepCopy())
- resourceTypes = append(resourceTypes, "ssl")
- case *apiv2.ApisixConsumer:
- result, err = d.translator.TranslateApisixConsumer(tctx,
t.DeepCopy())
- resourceTypes = append(resourceTypes, "consumer")
- case *v1alpha1.GatewayProxy:
- return d.updateConfigForGatewayProxy(tctx, t)
- }
- if err != nil {
- return err
- }
- if result == nil {
- return nil
- }
-
- oldParentRefs := d.getParentRefs(rk)
- if err := d.updateConfigs(rk, tctx); err != nil {
- return err
- }
- newParentRefs := d.getParentRefs(rk)
- deleteConfigs := d.findConfigsToDelete(oldParentRefs, newParentRefs)
- configs := d.getConfigs(rk)
-
- // sync delete
- if len(deleteConfigs) > 0 {
- err = d.sync(ctx, Task{
- Name: obj.GetName(),
- Labels: label.GenLabel(obj),
- ResourceTypes: resourceTypes,
- configs: deleteConfigs,
- })
- if err != nil {
- return err
- }
- for _, config := range deleteConfigs {
- if err := d.store.Delete(config.Name, resourceTypes,
label.GenLabel(obj)); err != nil {
- log.Errorw("failed to delete resources from
store",
- zap.String("name", config.Name),
- zap.Error(err),
- )
- return err
- }
- }
- }
-
- resources := adctypes.Resources{
- GlobalRules: result.GlobalRules,
- PluginMetadata: result.PluginMetadata,
- Services: result.Services,
- SSLs: result.SSL,
- Consumers: result.Consumers,
- }
- log.Debugw("update resources", zap.Any("resources", resources))
-
- for _, config := range configs {
- if err := d.store.Insert(config.Name, resourceTypes, resources,
label.GenLabel(obj)); err != nil {
- log.Errorw("failed to insert resources into store",
- zap.String("name", config.Name),
- zap.Error(err),
- )
- return err
- }
- }
-
- // This mode is full synchronization,
- // which only needs to be saved in cache
- // and triggered by a timer for synchronization
- if d.BackendMode == BackendModeAPISIXStandalone || d.BackendMode ==
BackendModeAPISIX {
- d.syncNotify()
- return nil
- }
-
- return d.sync(ctx, Task{
- Name: obj.GetName(),
- Labels: label.GenLabel(obj),
- Resources: resources,
- ResourceTypes: resourceTypes,
- configs: configs,
- })
-}
-
-func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
- log.Debugw("deleting object", zap.Any("object", obj))
-
- var resourceTypes []string
- var labels map[string]string
- switch obj.(type) {
- case *gatewayv1.HTTPRoute, *apiv2.ApisixRoute:
- resourceTypes = append(resourceTypes, "service")
- labels = label.GenLabel(obj)
- case *gatewayv1.Gateway:
- // delete all resources
- case *networkingv1.Ingress:
- resourceTypes = append(resourceTypes, "service", "ssl")
- labels = label.GenLabel(obj)
- case *v1alpha1.Consumer:
- resourceTypes = append(resourceTypes, "consumer")
- labels = label.GenLabel(obj)
- case *networkingv1.IngressClass:
- // delete all resources
- case *apiv2.ApisixGlobalRule:
- resourceTypes = append(resourceTypes, "global_rule")
- labels = label.GenLabel(obj)
- case *apiv2.ApisixTls:
- resourceTypes = append(resourceTypes, "ssl")
- labels = label.GenLabel(obj)
- case *apiv2.ApisixConsumer:
- resourceTypes = append(resourceTypes, "consumer")
- labels = label.GenLabel(obj)
- }
-
- rk := utils.NamespacedNameKind(obj)
-
- configs := d.getConfigs(rk)
- defer d.deleteConfigs(rk)
-
- for _, config := range configs {
- if err := d.store.Delete(config.Name, resourceTypes, labels);
err != nil {
- log.Errorw("failed to delete resources from store",
- zap.String("name", config.Name),
- zap.Error(err),
- )
- return err
- }
- }
-
- log.Debugw("successfully deleted resources from store",
zap.Any("object", obj))
-
- switch d.BackendMode {
- case BackendModeAPISIXStandalone, BackendModeAPISIX:
- // Full synchronization is performed on a gateway by gateway
basis
- // and it is not possible to perform scheduled synchronization
- // on deleted gateway level resources
- if len(resourceTypes) == 0 {
- return d.sync(ctx, Task{
- Name: obj.GetName(),
- configs: configs,
- })
- } else {
- d.syncNotify()
- }
- return nil
- default:
- log.Errorw("unknown backend mode", zap.String("mode",
d.BackendMode))
- return errors.New("unknown backend mode: " + d.BackendMode)
- }
-}
-
-func (d *adcClient) Start(ctx context.Context) error {
- d.readier.WaitReady(ctx, 5*time.Minute)
-
- initalSyncDelay := d.InitSyncDelay
- if initalSyncDelay > 0 {
- time.AfterFunc(initalSyncDelay, func() {
- if err := d.Sync(ctx); err != nil {
- log.Error(err)
- return
- }
- })
- }
-
- if d.SyncPeriod < 1 {
- return nil
- }
- ticker := time.NewTicker(d.SyncPeriod)
- defer ticker.Stop()
- for {
- synced := false
- select {
- case <-d.syncCh:
- synced = true
- case <-ticker.C:
- synced = true
- case <-ctx.Done():
- return nil
- }
- if synced {
- if err := d.Sync(ctx); err != nil {
- log.Error(err)
- }
- }
- }
-}
-
-func (d *adcClient) Sync(ctx context.Context) error {
- d.syncLock.Lock()
- defer d.syncLock.Unlock()
-
- log.Debug("syncing all resources")
-
- if len(d.configs) == 0 {
- return nil
- }
-
- cfg := map[string]adcConfig{}
- for _, config := range d.configs {
- cfg[config.Name] = config
- }
-
- log.Debugw("syncing resources with multiple configs",
zap.Any("configs", cfg))
-
- failedMap := map[string]types.ADCExecutionErrors{}
- var failedConfigs []string
- for name, config := range cfg {
- resources, err := d.store.GetResources(name)
- if err != nil {
- log.Errorw("failed to get resources from store",
zap.String("name", name), zap.Error(err))
- failedConfigs = append(failedConfigs, name)
- continue
- }
- if resources == nil {
- continue
- }
-
- if err := d.sync(ctx, Task{
- Name: name + "-sync",
- configs: []adcConfig{config},
- Resources: *resources,
- }); err != nil {
- log.Errorw("failed to sync resources",
zap.String("name", name), zap.Error(err))
- failedConfigs = append(failedConfigs, name)
- var execErrs types.ADCExecutionErrors
- if errors.As(err, &execErrs) {
- failedMap[name] = execErrs
- }
- }
- }
- d.handleADCExecutionErrors(failedMap)
- if len(failedConfigs) > 0 {
- return fmt.Errorf("failed to sync %d configs: %s",
- len(failedConfigs),
- strings.Join(failedConfigs, ", "))
- }
- return nil
-}
-
-func (d *adcClient) sync(ctx context.Context, task Task) error {
- log.Debugw("syncing resources", zap.Any("task", task))
-
- if len(task.configs) == 0 {
- log.Warnw("no adc configs provided", zap.Any("task", task))
- return nil
- }
-
- // Record file I/O duration
- fileIOStart := time.Now()
- syncFilePath, cleanup, err := prepareSyncFile(task.Resources)
- if err != nil {
- pkgmetrics.RecordFileIODuration("prepare_sync_file", "failure",
time.Since(fileIOStart).Seconds())
- return err
- }
- pkgmetrics.RecordFileIODuration("prepare_sync_file", "success",
time.Since(fileIOStart).Seconds())
- defer cleanup()
-
- args := BuildADCExecuteArgs(syncFilePath, task.Labels,
task.ResourceTypes)
-
- var errs types.ADCExecutionErrors
- for _, config := range task.configs {
- // Record sync duration for each config
- startTime := time.Now()
- resourceType := strings.Join(task.ResourceTypes, ",")
- if resourceType == "" {
- resourceType = "all"
- }
-
- err := d.executor.Execute(ctx, d.BackendMode, config, args)
- duration := time.Since(startTime).Seconds()
-
- status := "success"
- if err != nil {
- status = "failure"
- log.Errorw("failed to execute adc command",
zap.Error(err), zap.Any("config", config))
-
- var execErr types.ADCExecutionError
- if errors.As(err, &execErr) {
- errs.Errors = append(errs.Errors, execErr)
- pkgmetrics.RecordExecutionError(config.Name,
execErr.Name)
- } else {
- pkgmetrics.RecordExecutionError(config.Name,
"unknown")
- }
- }
-
- // Record metrics
- pkgmetrics.RecordSyncDuration(config.Name, resourceType,
status, duration)
- }
-
- if len(errs.Errors) > 0 {
- return errs
- }
- return nil
-}
-
-func (d *adcClient) syncNotify() {
- select {
- case d.syncCh <- struct{}{}:
- default:
- }
-}
-
-func prepareSyncFile(resources any) (string, func(), error) {
- data, err := json.Marshal(resources)
- if err != nil {
- return "", nil, err
- }
-
- tmpFile, err := os.CreateTemp("", "adc-task-*.json")
- if err != nil {
- return "", nil, err
- }
- cleanup := func() {
- _ = tmpFile.Close()
- _ = os.Remove(tmpFile.Name())
- }
- if _, err := tmpFile.Write(data); err != nil {
- cleanup()
- return "", nil, err
- }
-
- log.Debugw("generated adc file", zap.String("filename",
tmpFile.Name()), zap.String("json", string(data)))
-
- return tmpFile.Name(), cleanup, nil
-}
-
-func (d *adcClient) handleADCExecutionErrors(statusesMap
map[string]types.ADCExecutionErrors) {
- statusUpdateMap := d.resolveADCExecutionErrors(statusesMap)
- d.handleStatusUpdate(statusUpdateMap)
-}
-
-func (d *adcClient) NeedLeaderElection() bool {
- return true
-}
diff --git a/internal/provider/apisix/provider.go
b/internal/provider/apisix/provider.go
new file mode 100644
index 00000000..0246b3e7
--- /dev/null
+++ b/internal/provider/apisix/provider.go
@@ -0,0 +1,297 @@
+// 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"
+ "sync"
+ "time"
+
+ "github.com/api7/gopkg/pkg/log"
+ "go.uber.org/zap"
+ networkingv1 "k8s.io/api/networking/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"
+ "github.com/apache/apisix-ingress-controller/api/v1alpha1"
+ apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
+ adcclient
"github.com/apache/apisix-ingress-controller/internal/adc/client"
+ "github.com/apache/apisix-ingress-controller/internal/adc/translator"
+ "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/manager/readiness"
+ "github.com/apache/apisix-ingress-controller/internal/provider"
+ "github.com/apache/apisix-ingress-controller/internal/types"
+ "github.com/apache/apisix-ingress-controller/internal/utils"
+)
+
+const ProviderTypeAPISIX = "apisix"
+
+type apisixProvider struct {
+ provider.Options
+ sync.Mutex
+
+ translator *translator.Translator
+
+ updater status.Updater
+ statusUpdateMap map[types.NamespacedNameKind][]string
+
+ readier readiness.ReadinessManager
+
+ syncCh chan struct{}
+
+ client *adcclient.Client
+}
+
+func New(updater status.Updater, readier readiness.ReadinessManager, opts
...provider.Option) (provider.Provider, error) {
+ o := provider.Options{}
+ o.ApplyOptions(opts)
+ if o.BackendMode == "" {
+ o.BackendMode = ProviderTypeAPISIX
+ }
+
+ cli, err := adcclient.New(o.BackendMode, o.SyncTimeout)
+ if err != nil {
+ return nil, err
+ }
+
+ return &apisixProvider{
+ client: cli,
+ Options: o,
+ translator: &translator.Translator{},
+ updater: updater,
+ readier: readier,
+ syncCh: make(chan struct{}, 1),
+ }, nil
+}
+
+func (d *apisixProvider) Update(ctx context.Context, tctx
*provider.TranslateContext, obj client.Object) error {
+ log.Debugw("updating object", zap.Any("object", obj))
+ var (
+ result *translator.TranslateResult
+ resourceTypes []string
+ err error
+ )
+
+ rk := utils.NamespacedNameKind(obj)
+
+ switch t := obj.(type) {
+ case *gatewayv1.HTTPRoute:
+ result, err = d.translator.TranslateHTTPRoute(tctx,
t.DeepCopy())
+ resourceTypes = append(resourceTypes, "service")
+ case *gatewayv1.Gateway:
+ result, err = d.translator.TranslateGateway(tctx, t.DeepCopy())
+ resourceTypes = append(resourceTypes, "global_rule", "ssl",
"plugin_metadata")
+ case *networkingv1.Ingress:
+ result, err = d.translator.TranslateIngress(tctx, t.DeepCopy())
+ resourceTypes = append(resourceTypes, "service", "ssl")
+ case *v1alpha1.Consumer:
+ result, err = d.translator.TranslateConsumerV1alpha1(tctx,
t.DeepCopy())
+ resourceTypes = append(resourceTypes, "consumer")
+ case *networkingv1.IngressClass:
+ result, err = d.translator.TranslateIngressClass(tctx,
t.DeepCopy())
+ resourceTypes = append(resourceTypes, "global_rule",
"plugin_metadata")
+ case *apiv2.ApisixRoute:
+ result, err = d.translator.TranslateApisixRoute(tctx,
t.DeepCopy())
+ resourceTypes = append(resourceTypes, "service")
+ case *apiv2.ApisixGlobalRule:
+ result, err = d.translator.TranslateApisixGlobalRule(tctx,
t.DeepCopy())
+ resourceTypes = append(resourceTypes, "global_rule")
+ case *apiv2.ApisixTls:
+ result, err = d.translator.TranslateApisixTls(tctx,
t.DeepCopy())
+ resourceTypes = append(resourceTypes, "ssl")
+ case *apiv2.ApisixConsumer:
+ result, err = d.translator.TranslateApisixConsumer(tctx,
t.DeepCopy())
+ resourceTypes = append(resourceTypes, "consumer")
+ case *v1alpha1.GatewayProxy:
+ return d.updateConfigForGatewayProxy(tctx, t)
+ }
+ if err != nil {
+ return err
+ }
+ if result == nil {
+ return nil
+ }
+
+ configs, err := d.buildConfig(tctx, rk)
+ if err != nil {
+ return err
+ }
+
+ if len(configs) == 0 {
+ return nil
+ }
+
+ defer d.syncNotify()
+
+ return d.client.UpdateConfig(ctx, adcclient.Task{
+ Key: rk,
+ Name: rk.String(),
+ Labels: label.GenLabel(obj),
+ Configs: configs,
+ ResourceTypes: resourceTypes,
+ Resources: &adctypes.Resources{
+ GlobalRules: result.GlobalRules,
+ PluginMetadata: result.PluginMetadata,
+ Services: result.Services,
+ SSLs: result.SSL,
+ Consumers: result.Consumers,
+ },
+ })
+}
+
+func (d *apisixProvider) Delete(ctx context.Context, obj client.Object) error {
+ log.Debugw("deleting object", zap.Any("object", obj))
+
+ var resourceTypes []string
+ var labels map[string]string
+ switch obj.(type) {
+ case *gatewayv1.HTTPRoute, *apiv2.ApisixRoute:
+ resourceTypes = append(resourceTypes, "service")
+ labels = label.GenLabel(obj)
+ case *gatewayv1.Gateway:
+ // delete all resources
+ case *networkingv1.Ingress:
+ resourceTypes = append(resourceTypes, "service", "ssl")
+ labels = label.GenLabel(obj)
+ case *v1alpha1.Consumer:
+ resourceTypes = append(resourceTypes, "consumer")
+ labels = label.GenLabel(obj)
+ case *networkingv1.IngressClass:
+ // delete all resources
+ case *apiv2.ApisixGlobalRule:
+ resourceTypes = append(resourceTypes, "global_rule")
+ labels = label.GenLabel(obj)
+ case *apiv2.ApisixTls:
+ resourceTypes = append(resourceTypes, "ssl")
+ labels = label.GenLabel(obj)
+ case *apiv2.ApisixConsumer:
+ resourceTypes = append(resourceTypes, "consumer")
+ labels = label.GenLabel(obj)
+ }
+ nnk := utils.NamespacedNameKind(obj)
+
+ // Full synchronization is performed on a gateway by gateway basis
+ // and it is not possible to perform scheduled synchronization
+ // on deleted gateway level resources
+ if len(resourceTypes) == 0 {
+ return d.client.Delete(ctx, adcclient.Task{
+ Key: nnk,
+ Name: nnk.String(),
+ Labels: labels,
+ })
+ }
+ defer d.syncNotify()
+ return d.client.DeleteConfig(ctx, adcclient.Task{
+ Key: nnk,
+ Name: nnk.String(),
+ Labels: labels,
+ ResourceTypes: resourceTypes,
+ })
+}
+
+func (d *apisixProvider) buildConfig(tctx *provider.TranslateContext, nnk
types.NamespacedNameKind) (map[types.NamespacedNameKind]adctypes.Config, error)
{
+ configs := make(map[types.NamespacedNameKind]adctypes.Config,
len(tctx.ResourceParentRefs[nnk]))
+ for _, gp := range tctx.GatewayProxies {
+ config, err := d.translator.TranslateGatewayProxyToConfig(tctx,
&gp, d.ResolveEndpoints)
+ if err != nil {
+ return nil, err
+ }
+ configs[utils.NamespacedNameKind(&gp)] = *config
+ }
+ return configs, nil
+}
+
+func (d *apisixProvider) Start(ctx context.Context) error {
+ d.readier.WaitReady(ctx, 5*time.Minute)
+
+ initalSyncDelay := d.InitSyncDelay
+ if initalSyncDelay > 0 {
+ time.AfterFunc(initalSyncDelay, func() {
+ if err := d.sync(ctx); err != nil {
+ log.Error(err)
+ return
+ }
+ })
+ }
+
+ if d.SyncPeriod < 1 {
+ return nil
+ }
+ ticker := time.NewTicker(d.SyncPeriod)
+ defer ticker.Stop()
+ for {
+ synced := false
+ select {
+ case <-d.syncCh:
+ synced = true
+ case <-ticker.C:
+ synced = true
+ case <-ctx.Done():
+ return nil
+ }
+ if synced {
+ if err := d.sync(ctx); err != nil {
+ log.Error(err)
+ }
+ }
+ }
+}
+
+func (d *apisixProvider) sync(ctx context.Context) error {
+ statusesMap, err := d.client.Sync(ctx)
+ d.handleADCExecutionErrors(statusesMap)
+ return err
+}
+
+func (d *apisixProvider) syncNotify() {
+ select {
+ case d.syncCh <- struct{}{}:
+ default:
+ }
+}
+
+func (d *apisixProvider) handleADCExecutionErrors(statusesMap
map[string]types.ADCExecutionErrors) {
+ statusUpdateMap := d.resolveADCExecutionErrors(statusesMap)
+ d.handleStatusUpdate(statusUpdateMap)
+ log.Debugw("handled ADC execution errors", zap.Any("status_record",
statusesMap), zap.Any("status_update", statusUpdateMap))
+}
+
+func (d *apisixProvider) NeedLeaderElection() bool {
+ return true
+}
+
+// updateConfigForGatewayProxy update config for all referrers of the
GatewayProxy
+func (d *apisixProvider) updateConfigForGatewayProxy(tctx
*provider.TranslateContext, gp *v1alpha1.GatewayProxy) error {
+ config, err := d.translator.TranslateGatewayProxyToConfig(tctx, gp,
d.ResolveEndpoints)
+ if err != nil {
+ return err
+ }
+
+ nnk := utils.NamespacedNameKind(gp)
+ if config == nil {
+ d.client.ConfigManager.DeleteConfig(nnk)
+ return nil
+ }
+ referrers := tctx.GatewayProxyReferrers[utils.NamespacedName(gp)]
+ d.client.ConfigManager.SetConfigRefs(nnk, referrers)
+ d.client.ConfigManager.UpdateConfig(nnk, *config)
+ d.syncNotify()
+ return nil
+}
diff --git a/internal/provider/adc/status.go
b/internal/provider/apisix/status.go
similarity index 90%
rename from internal/provider/adc/status.go
rename to internal/provider/apisix/status.go
index 8d1f70f8..838fd715 100644
--- a/internal/provider/adc/status.go
+++ b/internal/provider/apisix/status.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package adc
+package apisix
import (
"fmt"
@@ -41,7 +41,7 @@ import (
// 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 *adcClient) handleStatusUpdate(statusUpdateMap
map[types.NamespacedNameKind][]string) {
+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(
@@ -67,7 +67,7 @@ func (d *adcClient) handleStatusUpdate(statusUpdateMap
map[types.NamespacedNameK
d.statusUpdateMap = statusUpdateMap
}
-func (d *adcClient) updateStatus(nnk types.NamespacedNameKind, condition
metav1.Condition) {
+func (d *apisixProvider) updateStatus(nnk types.NamespacedNameKind, condition
metav1.Condition) {
switch nnk.Kind {
case types.KindApisixRoute:
d.updater.Update(status.Update{
@@ -110,7 +110,8 @@ func (d *adcClient) updateStatus(nnk
types.NamespacedNameKind, condition metav1.
}),
})
case types.KindHTTPRoute:
- parentRefs := d.getParentRefs(nnk)
+ parentRefs :=
d.client.ConfigManager.GetConfigRefsByResourceKey(nnk)
+ log.Debugw("updating HTTPRoute status", zap.Any("parentRefs",
parentRefs))
gatewayRefs := map[types.NamespacedNameKind]struct{}{}
for _, parentRef := range parentRefs {
if parentRef.Kind == types.KindGateway {
@@ -146,7 +147,7 @@ func (d *adcClient) updateStatus(nnk
types.NamespacedNameKind, condition metav1.
}
}
-func (d *adcClient) resolveADCExecutionErrors(
+func (d *apisixProvider) resolveADCExecutionErrors(
statusesMap map[string]types.ADCExecutionErrors,
) map[types.NamespacedNameKind][]string {
statusUpdateMap := map[types.NamespacedNameKind][]string{}
@@ -165,12 +166,12 @@ func (d *adcClient) resolveADCExecutionErrors(
return statusUpdateMap
}
-func (d *adcClient) handleEmptyFailedStatuses(
+func (d *apisixProvider) handleEmptyFailedStatuses(
configName string,
failedStatus types.ADCExecutionServerAddrError,
statusUpdateMap map[types.NamespacedNameKind][]string,
) {
- resource, err := d.store.GetResources(configName)
+ resource, err := d.client.GetResources(configName)
if err != nil {
log.Errorw("failed to get resources from store",
zap.String("configName", configName), zap.Error(err))
return
@@ -188,7 +189,7 @@ func (d *adcClient) handleEmptyFailedStatuses(
d.addResourceToStatusUpdateMap(obj.GetLabels(),
failedStatus.Error(), statusUpdateMap)
}
- globalRules, err := d.store.ListGlobalRules(configName)
+ globalRules, err := d.client.ListGlobalRules(configName)
if err != nil {
log.Errorw("failed to list global rules",
zap.String("configName", configName), zap.Error(err))
return
@@ -198,14 +199,14 @@ func (d *adcClient) handleEmptyFailedStatuses(
}
}
-func (d *adcClient) handleDetailedFailedStatuses(
+func (d *apisixProvider) handleDetailedFailedStatuses(
configName string,
failedStatus types.ADCExecutionServerAddrError,
statusUpdateMap map[types.NamespacedNameKind][]string,
) {
for _, status := range failedStatus.FailedStatuses {
id := status.Event.ResourceID
- labels, err := d.store.GetResourceLabel(configName,
status.Event.ResourceType, id)
+ labels, err := d.client.GetResourceLabel(configName,
status.Event.ResourceType, id)
if err != nil {
log.Errorw("failed to get resource label",
zap.String("configName", configName),
@@ -223,7 +224,7 @@ func (d *adcClient) handleDetailedFailedStatuses(
}
}
-func (d *adcClient) addResourceToStatusUpdateMap(
+func (d *apisixProvider) addResourceToStatusUpdateMap(
labels map[string]string,
msg string,
statusUpdateMap map[types.NamespacedNameKind][]string,
diff --git a/internal/provider/common/configmanager.go
b/internal/provider/common/configmanager.go
new file mode 100644
index 00000000..387537b0
--- /dev/null
+++ b/internal/provider/common/configmanager.go
@@ -0,0 +1,170 @@
+// 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 common
+
+import (
+ "sync"
+)
+
+/*
+ConfigManager is a generic configuration association manager that maintains
relationships between:
+ - Resource objects (e.g., HTTPRoute, Ingress)
+ - Configuration objects (e.g., Gateway, IngressClass configurations)
+ - Referenced resources (e.g., Secrets referenced by configurations)
+
+Core data structure:
+ - resourceConfigKeys: Resource object → Configuration keys
+ (e.g., HTTPRoute/Ingress → [ConfigKey1, ConfigKey2])
+ - configs: Configuration storage pool
+ (e.g., ConfigKey → GatewayProxy Configuration)
+ - configRefs: Configuration → Referenced resources
+ (e.g., ConfigKey → [Gateway, IngressClass])
+
+Main relationships:
+ Resource objects (HTTPRoute/Ingress) --resourceConfigKeys--> Configuration
objects
+ --configRefs--> Referenced resources (Gateways/IngressClasses)
+
+Note: The referenced resources in configRefs are typically higher-level
abstractions
+that configurations depend on (Gateway/IngressClass), not the low-level
resources like Secrets.
+*/
+
+type ConfigManager[K comparable, T any] struct {
+ mu sync.Mutex
+
+ configs map[K]T
+ configRefs map[K][]K
+
+ resourceConfigKeys map[K][]K
+}
+
+func NewConfigManager[K comparable, T any]() *ConfigManager[K, T] {
+ return &ConfigManager[K, T]{
+ resourceConfigKeys: make(map[K][]K),
+ configs: make(map[K]T),
+ configRefs: make(map[K][]K),
+ }
+}
+
+func (s *ConfigManager[K, T]) GetConfigRefs(key K) []K {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.configRefs[key]
+}
+
+func (s *ConfigManager[K, T]) GetConfigRefsByResourceKey(key K) []K {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ configKeys, ok := s.resourceConfigKeys[key]
+ if !ok {
+ return nil
+ }
+ refs := make([]K, 0, len(configKeys))
+ for _, k := range configKeys {
+ if ref, ok := s.configRefs[k]; ok {
+ refs = append(refs, ref...)
+ }
+ }
+ return refs
+}
+
+func (s *ConfigManager[K, T]) SetConfigRefs(key K, refs []K) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.configRefs[key] = refs
+}
+
+func (s *ConfigManager[K, T]) Get(key K) map[K]T {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ resourceConfigKeys := s.resourceConfigKeys[key]
+ configs := make(map[K]T, len(resourceConfigKeys))
+ for _, parent := range resourceConfigKeys {
+ if cfg, ok := s.configs[parent]; ok {
+ configs[parent] = cfg
+ }
+ }
+ return configs
+}
+
+func (s *ConfigManager[K, T]) List() map[K]T {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ configs := make(map[K]T, len(s.configs))
+ for k, v := range s.configs {
+ configs[k] = v
+ }
+ return configs
+}
+
+func (s *ConfigManager[K, T]) UpdateConfig(key K, cfg T) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.configs[key] = cfg
+}
+
+func (s *ConfigManager[K, T]) Update(
+ key K,
+ mapRefs map[K]T,
+) (discard map[K]T) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ parentRefSet := make(map[K]struct{})
+ oldParentRefs := s.resourceConfigKeys[key]
+ newRefs := make([]K, 0, len(mapRefs))
+
+ for k, v := range mapRefs {
+ newRefs = append(newRefs, k)
+ s.configs[k] = v
+ parentRefSet[k] = struct{}{}
+ }
+ s.resourceConfigKeys[key] = newRefs
+ discard = make(map[K]T)
+ for _, old := range oldParentRefs {
+ if _, stillUsed := parentRefSet[old]; !stillUsed {
+ if cfg, ok := s.configs[old]; ok {
+ discard[old] = cfg
+ }
+ }
+ }
+
+ return discard
+}
+
+func (s *ConfigManager[K, T]) Set(key K, cfg T) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.configs[key] = cfg
+}
+
+func (s *ConfigManager[K, T]) Delete(key K) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ delete(s.resourceConfigKeys, key)
+ delete(s.configs, key)
+ delete(s.configRefs, key)
+}
+
+func (s *ConfigManager[K, T]) DeleteConfig(key K) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ delete(s.configs, key)
+ delete(s.configRefs, key)
+}
diff --git a/internal/provider/adc/translator/translator.go
b/internal/provider/init/init.go
similarity index 53%
rename from internal/provider/adc/translator/translator.go
rename to internal/provider/init/init.go
index 67ab7596..3fc68912 100644
--- a/internal/provider/adc/translator/translator.go
+++ b/internal/provider/init/init.go
@@ -15,22 +15,20 @@
// specific language governing permissions and limitations
// under the License.
-package translator
+package init
import (
- "github.com/go-logr/logr"
-
- adctypes "github.com/apache/apisix-ingress-controller/api/adc"
+ "github.com/apache/apisix-ingress-controller/internal/controller/status"
+ "github.com/apache/apisix-ingress-controller/internal/manager/readiness"
+ "github.com/apache/apisix-ingress-controller/internal/provider"
+ "github.com/apache/apisix-ingress-controller/internal/provider/apisix"
)
-type Translator struct {
- Log logr.Logger
-}
-type TranslateResult struct {
- Routes []*adctypes.Route
- Services []*adctypes.Service
- SSL []*adctypes.SSL
- GlobalRules adctypes.GlobalRule
- PluginMetadata adctypes.PluginMetadata
- Consumers []*adctypes.Consumer
+func init() {
+ provider.Register("apisix", apisix.New)
+ provider.Register("apisix-standalone", func(statusUpdater
status.Updater, readinessManager readiness.ReadinessManager, opts
...provider.Option) (provider.Provider, error) {
+ opts = append(opts,
provider.WithBackendMode("apisix-standalone"))
+ opts = append(opts, provider.WithResolveEndpoints())
+ return apisix.New(statusUpdater, readinessManager, opts...)
+ })
}
diff --git a/internal/provider/adc/options.go b/internal/provider/options.go
similarity index 67%
rename from internal/provider/adc/options.go
rename to internal/provider/options.go
index 8b6d7631..540f2e63 100644
--- a/internal/provider/adc/options.go
+++ b/internal/provider/options.go
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-package adc
+package provider
import "time"
@@ -24,10 +24,11 @@ type Option interface {
}
type Options struct {
- SyncTimeout time.Duration
- SyncPeriod time.Duration
- InitSyncDelay time.Duration
- BackendMode string
+ SyncTimeout time.Duration
+ SyncPeriod time.Duration
+ InitSyncDelay time.Duration
+ BackendMode string
+ ResolveEndpoints bool
}
func (o *Options) ApplyToList(lo *Options) {
@@ -43,6 +44,9 @@ func (o *Options) ApplyToList(lo *Options) {
if o.BackendMode != "" {
lo.BackendMode = o.BackendMode
}
+ if o.ResolveEndpoints {
+ lo.ResolveEndpoints = o.ResolveEndpoints
+ }
}
func (o *Options) ApplyOptions(opts []Option) *Options {
@@ -51,3 +55,23 @@ func (o *Options) ApplyOptions(opts []Option) *Options {
}
return o
}
+
+type backendModeOption string
+
+func (b backendModeOption) ApplyToList(o *Options) {
+ o.BackendMode = string(b)
+}
+
+func WithBackendMode(mode string) Option {
+ return backendModeOption(mode)
+}
+
+type resolveEndpointsOption bool
+
+func (r resolveEndpointsOption) ApplyToList(o *Options) {
+ o.ResolveEndpoints = bool(r)
+}
+
+func WithResolveEndpoints() Option {
+ return resolveEndpointsOption(true)
+}
diff --git a/internal/provider/provider.go b/internal/provider/provider.go
index d9b6cbfa..ef93de54 100644
--- a/internal/provider/provider.go
+++ b/internal/provider/provider.go
@@ -35,7 +35,6 @@ import (
type Provider interface {
Update(context.Context, *TranslateContext, client.Object) error
Delete(context.Context, client.Object) error
- Sync(context.Context) error
Start(context.Context) error
NeedLeaderElection() bool
}
diff --git a/internal/provider/register.go b/internal/provider/register.go
new file mode 100644
index 00000000..a2542ad7
--- /dev/null
+++ b/internal/provider/register.go
@@ -0,0 +1,49 @@
+// 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 provider
+
+import (
+ "fmt"
+
+ "github.com/apache/apisix-ingress-controller/internal/controller/status"
+ "github.com/apache/apisix-ingress-controller/internal/manager/readiness"
+)
+
+type RegisterFunc func(status.Updater, readiness.ReadinessManager, ...Option)
(Provider, error)
+
+var providers = map[string]RegisterFunc{}
+
+func Register(name string, registerFunc RegisterFunc) {
+ providers[name] = registerFunc
+}
+
+func Get(name string) (RegisterFunc, error) {
+ f, ok := providers[name]
+ if !ok {
+ return nil, fmt.Errorf("provider %q not found", name)
+ }
+ return f, nil
+}
+
+func New(providerType string, updater status.Updater, readinesser
readiness.ReadinessManager, opts ...Option) (Provider, error) {
+ f, err := Get(providerType)
+ if err != nil {
+ return nil, fmt.Errorf("failed to get provider %q: %w",
providerType, err)
+ }
+ return f(updater, readinesser, opts...)
+}
diff --git a/internal/types/types.go b/internal/types/types.go
index f695da87..5fe4b6a2 100644
--- a/internal/types/types.go
+++ b/internal/types/types.go
@@ -18,6 +18,9 @@
package types
import (
+ "fmt"
+ "strings"
+
k8stypes "k8s.io/apimachinery/pkg/types"
)
@@ -27,9 +30,33 @@ type NamespacedNameKind struct {
Kind string
}
-func (n *NamespacedNameKind) NamespacedName() k8stypes.NamespacedName {
+func (n NamespacedNameKind) NamespacedName() k8stypes.NamespacedName {
return k8stypes.NamespacedName{
Namespace: n.Namespace,
Name: n.Name,
}
}
+
+func (n NamespacedNameKind) String() string {
+ return n.Kind + "/" + n.Namespace + "/" + n.Name
+}
+
+func (n NamespacedNameKind) MarshalText() ([]byte, error) {
+ return []byte(n.String()), nil
+}
+
+func (n *NamespacedNameKind) UnmarshalText(text []byte) error {
+ return n.FromString(string(text))
+}
+
+func (n *NamespacedNameKind) FromString(s string) error {
+ parts := strings.Split(s, "/")
+ if len(parts) != 3 {
+ return fmt.Errorf("invalid format for NamespacedNameKind: %q,
expected Kind/Namespace/Name", s)
+ }
+
+ n.Kind = parts[0]
+ n.Namespace = parts[1]
+ n.Name = parts[2]
+ return nil
+}
diff --git a/test/e2e/crds/v1alpha1/gatewayproxy.go
b/test/e2e/crds/v1alpha1/gatewayproxy.go
index 7668cfa1..abe56aac 100644
--- a/test/e2e/crds/v1alpha1/gatewayproxy.go
+++ b/test/e2e/crds/v1alpha1/gatewayproxy.go
@@ -30,7 +30,6 @@ import (
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/utils/ptr"
- "github.com/apache/apisix-ingress-controller/internal/provider/adc"
"github.com/apache/apisix-ingress-controller/test/e2e/framework"
"github.com/apache/apisix-ingress-controller/test/e2e/scaffold"
)
@@ -187,7 +186,7 @@ spec:
keyword string
)
- if framework.ProviderType == adc.BackendModeAPISIX {
+ if framework.ProviderType ==
framework.ProviderTypeAPISIX {
keyword = fmt.Sprintf(`{"config.ServerAddrs":
["%s"]}`, s.Deployer.GetAdminEndpoint())
} else {
keyword = fmt.Sprintf(`{"config.ServerAddrs":
["http://%s:9180"]}`, s.GetPodIP(s.Namespace(),
"app.kubernetes.io/name=apisix"))
diff --git a/test/e2e/framework/apisix_consts.go
b/test/e2e/framework/apisix_consts.go
index 42214982..1761b1ec 100644
--- a/test/e2e/framework/apisix_consts.go
+++ b/test/e2e/framework/apisix_consts.go
@@ -30,6 +30,11 @@ var (
ProviderType = cmp.Or(os.Getenv("PROVIDER_TYPE"), "apisix")
)
+const (
+ ProviderTypeAPISIX = "apisix"
+ ProviderTypeAPISIXStandalone = "apisix-standalone"
+)
+
var (
//go:embed manifests/apisix.yaml
apisixStandaloneTemplate string
diff --git a/test/e2e/framework/manifests/apisix-standalone.yaml
b/test/e2e/framework/manifests/apisix-standalone.yaml
index 457a68b1..1a76b29c 100644
--- a/test/e2e/framework/manifests/apisix-standalone.yaml
+++ b/test/e2e/framework/manifests/apisix-standalone.yaml
@@ -54,7 +54,7 @@ spec:
spec:
initContainers:
- name: config-setup
- image: apache/apisix:dev
+ image: apache/apisix:3.13.0-ubuntu
command:
- sh
- -c
@@ -72,7 +72,7 @@ spec:
mountPath: /tmp/apisix-conf
containers:
- name: apisix
- image: apache/apisix:dev
+ image: apache/apisix:3.13.0-ubuntu
ports:
- name: http
containerPort: 9080
diff --git a/test/e2e/framework/manifests/apisix.yaml
b/test/e2e/framework/manifests/apisix.yaml
index b01beca1..3117ccee 100644
--- a/test/e2e/framework/manifests/apisix.yaml
+++ b/test/e2e/framework/manifests/apisix.yaml
@@ -61,7 +61,7 @@ spec:
spec:
initContainers:
- name: config-setup
- image: apache/apisix:dev
+ image: apache/apisix:3.13.0-ubuntu
command:
- sh
- -c
@@ -79,7 +79,7 @@ spec:
mountPath: /tmp/apisix-conf
containers:
- name: apisix
- image: apache/apisix:dev
+ image: apache/apisix:3.13.0-ubuntu
ports:
- name: http
containerPort: 9080
diff --git a/test/e2e/scaffold/adc.go b/test/e2e/scaffold/adc.go
index 12bd229e..11d5a712 100644
--- a/test/e2e/scaffold/adc.go
+++ b/test/e2e/scaffold/adc.go
@@ -30,7 +30,7 @@ import (
"gopkg.in/yaml.v3"
adctypes "github.com/apache/apisix-ingress-controller/api/adc"
-
"github.com/apache/apisix-ingress-controller/internal/provider/adc/translator"
+ "github.com/apache/apisix-ingress-controller/internal/adc/translator"
)
// DataplaneResource defines the interface for accessing dataplane resources
diff --git a/test/e2e/scaffold/apisix_deployer.go
b/test/e2e/scaffold/apisix_deployer.go
index 740f9e02..f83cd29d 100644
--- a/test/e2e/scaffold/apisix_deployer.go
+++ b/test/e2e/scaffold/apisix_deployer.go
@@ -30,7 +30,6 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/utils/ptr"
- "github.com/apache/apisix-ingress-controller/internal/provider/adc"
"github.com/apache/apisix-ingress-controller/pkg/utils"
"github.com/apache/apisix-ingress-controller/test/e2e/framework"
)
@@ -215,7 +214,7 @@ func (s *APISIXDeployer) deployDataplane(opts
*APISIXDeployOptions) *corev1.Serv
kubectlOpts := k8s.NewKubectlOptions("", "", opts.Namespace)
- if framework.ProviderType == adc.BackendModeAPISIX {
+ if framework.ProviderType == framework.ProviderTypeAPISIX {
opts.ConfigProvider = "etcd"
// deploy etcd
k8s.KubectlApplyFromString(s.GinkgoT, kubectlOpts,
framework.EtcdSpec)
