This is an automated email from the ASF dual-hosted git repository.
kvn 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 121bda3 feat: use unified translator to translate ApisixRoute v1
(#265)
121bda3 is described below
commit 121bda31cbe20c39b0b2182257278783320e71c4
Author: Alex Zhang <[email protected]>
AuthorDate: Thu Feb 25 10:32:26 2021 +0800
feat: use unified translator to translate ApisixRoute v1 (#265)
* feat: use unified translator to translage ApisixRoute v1
* fix: broken e2e test cases
---
pkg/ingress/apisix/plugin.go | 52 --------
pkg/ingress/apisix/route.go | 146 --------------------
pkg/ingress/apisix/route_test.go | 64 ---------
pkg/ingress/apisix/tls.go | 4 -
pkg/ingress/apisix/tls_test.go | 40 ------
pkg/ingress/controller/apisix_route.go | 34 +++--
.../translation/annotations.go} | 63 +++++----
pkg/kube/translation/apisix_route.go | 86 ++++++++++++
pkg/kube/translation/translator.go | 5 +-
pkg/seven/state/builder.go | 35 ++---
pkg/seven/state/route_worker.go | 4 +-
pkg/seven/state/service_worker.go | 147 ---------------------
pkg/seven/state/solver.go | 14 +-
13 files changed, 162 insertions(+), 532 deletions(-)
diff --git a/pkg/ingress/apisix/plugin.go b/pkg/ingress/apisix/plugin.go
deleted file mode 100644
index 20fe081..0000000
--- a/pkg/ingress/apisix/plugin.go
+++ /dev/null
@@ -1,52 +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 apisix
-
-import (
- "strconv"
-
- "github.com/apache/apisix-ingress-controller/pkg/seven/apisix"
-)
-
-type CorsYaml struct {
- Enable bool `json:"enable,omitempty"`
- AllowOrigin string `json:"allow_origin,omitempty"`
- AllowHeaders string `json:"allow_headers,omitempty"`
- AllowMethods string `json:"allow_methods,omitempty"`
-}
-
-func (c *CorsYaml) SetEnable(enable string) {
- if b, err := strconv.ParseBool(enable); err != nil {
- c.Enable = false
- } else {
- c.Enable = b
- }
-}
-
-func (c *CorsYaml) SetOrigin(origin string) {
- c.AllowOrigin = origin
-}
-
-func (c *CorsYaml) SetHeaders(headers string) {
- c.AllowHeaders = headers
-}
-func (c *CorsYaml) SetMethods(methods string) {
- c.AllowMethods = methods
-}
-
-func (c *CorsYaml) Build() *apisix.Cors {
- maxAge := int64(3600)
- return apisix.BuildCors(c.Enable, &c.AllowOrigin, &c.AllowHeaders,
&c.AllowMethods, &maxAge)
-}
diff --git a/pkg/ingress/apisix/route.go b/pkg/ingress/apisix/route.go
deleted file mode 100644
index be890ba..0000000
--- a/pkg/ingress/apisix/route.go
+++ /dev/null
@@ -1,146 +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 apisix
-
-import (
- "strconv"
-
- configv1
"github.com/apache/apisix-ingress-controller/pkg/kube/apisix/apis/config/v1"
- "github.com/apache/apisix-ingress-controller/pkg/kube/translation"
- "github.com/apache/apisix-ingress-controller/pkg/seven/conf"
- apisix "github.com/apache/apisix-ingress-controller/pkg/types/apisix/v1"
-)
-
-const (
- DefaultLBType = "roundrobin"
- SSLREDIRECT = "k8s.apisix.apache.org/ssl-redirect"
- WHITELIST = "k8s.apisix.apache.org/whitelist-source-range"
- ENABLE_CORS = "k8s.apisix.apache.org/enable-cors"
- CORS_ALLOW_ORIGIN = "k8s.apisix.apache.org/cors-allow-origin"
- CORS_ALLOW_HEADERS = "k8s.apisix.apache.org/cors-allow-headers"
- CORS_ALLOW_METHODS = "k8s.apisix.apache.org/cors-allow-methods"
- INGRESS_CLASS = "k8s.apisix.apache.org/ingress.class"
-)
-
-type ApisixRoute configv1.ApisixRoute
-
-// Convert convert to apisix.Route from ingress.ApisixRoute CRD
-func (ar *ApisixRoute) Convert(translator translation.Translator)
([]*apisix.Route, []*apisix.Service, []*apisix.Upstream, error) {
- ns := ar.Namespace
- // meta annotation
- plugins, group := BuildAnnotation(ar.Annotations)
- conf.AddGroup(group)
- // Host
- rules := ar.Spec.Rules
- routes := make([]*apisix.Route, 0)
- services := make([]*apisix.Service, 0)
- serviceMap := make(map[string]*apisix.Service)
- upstreams := make([]*apisix.Upstream, 0)
- upstreamMap := make(map[string]*apisix.Upstream)
- rv := ar.ObjectMeta.ResourceVersion
- for _, r := range rules {
- host := r.Host
- paths := r.Http.Paths
- for _, p := range paths {
- uri := p.Path
- svcName := p.Backend.ServiceName
- svcPort := strconv.Itoa(p.Backend.ServicePort)
- // apisix route name = host + path
- apisixRouteName := host + uri
- // apisix service name = namespace_svcName_svcPort
- apisixSvcName := ns + "_" + svcName + "_" + svcPort
- // apisix route name = namespace_svcName_svcPort =
apisix service name
- apisixUpstreamName := ns + "_" + svcName + "_" + svcPort
- // plugins defined in Route Level
- pls := p.Plugins
- pluginRet := make(apisix.Plugins)
- // 1.add annotation plugins
- for k, v := range plugins {
- pluginRet[k] = v
- }
- // 2.add route plugins
- for _, p := range pls {
- if p.Enable {
- if p.Config != nil {
- pluginRet[p.Name] = p.Config
- } else if p.ConfigSet != nil {
- pluginRet[p.Name] = p.ConfigSet
- } else {
- pluginRet[p.Name] =
make(map[string]interface{})
- }
- }
- }
- // fullRouteName
- fullRouteName := apisixRouteName
- if group != "" {
- fullRouteName = group + "_" + apisixRouteName
- }
-
- // routes
- route := &apisix.Route{
- Metadata: apisix.Metadata{
- Group: group,
- FullName: fullRouteName,
- ResourceVersion: rv,
- Name: apisixRouteName,
- },
- Host: host,
- Path: uri,
- ServiceName: apisixSvcName,
- UpstreamName: apisixUpstreamName,
- Plugins: pluginRet,
- }
- routes = append(routes, route)
- // services
- // fullServiceName
- fullServiceName := apisixSvcName
- if group != "" {
- fullServiceName = group + "_" + apisixSvcName
- }
-
- service := &apisix.Service{
- FullName: fullServiceName,
- Group: group,
- Name: apisixSvcName,
- UpstreamName: apisixUpstreamName,
- ResourceVersion: rv,
- }
- serviceMap[service.FullName] = service
-
- // upstreams
- // fullServiceName
- fullUpstreamName := apisixUpstreamName
- if group != "" {
- fullUpstreamName = group + "_" +
apisixUpstreamName
- }
- ups, err := translator.TranslateUpstream(ns, svcName,
int32(p.Backend.ServicePort))
- if err != nil {
- return nil, nil, nil, err
- }
- ups.FullName = fullUpstreamName
- ups.Group = group
- ups.ResourceVersion = rv
- ups.Name = apisixUpstreamName
- upstreamMap[ups.FullName] = ups
- }
- }
- for _, s := range serviceMap {
- services = append(services, s)
- }
- for _, u := range upstreamMap {
- upstreams = append(upstreams, u)
- }
- return routes, services, upstreams, nil
-}
diff --git a/pkg/ingress/apisix/route_test.go b/pkg/ingress/apisix/route_test.go
deleted file mode 100644
index bf0e1a2..0000000
--- a/pkg/ingress/apisix/route_test.go
+++ /dev/null
@@ -1,64 +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 apisix
-
-import "testing"
-
-func TestApisixRoute_Convert(t *testing.T) {
-
-}
-
-var routeYaml = `
-apiVersion: apisix.apache.org/v1
-kind: ApisixRoute
-metadata:
- annotations:
- k8s.apisix.apache.org/cors-allow-headers:
DNT,X-CustomHeader,Keep-Alive,User-Agent,X-Requested-With,If-Modified-Since,Cache-Control,Content-Type,Authorization,openID,audiotoken
- k8s.apisix.apache.org/cors-allow-methods: HEAD,GET,POST,PUT,PATCH,DELETE
- k8s.apisix.apache.org/cors-allow-origin: '*'
- k8s.apisix.apache.org/enable-cors: "true"
- k8s.apisix.apache.org/ssl-redirect: "false"
- k8s.apisix.apache.org/whitelist-source-range: 58.210.212.110,10.244.0.0/16
- name: httpserver-route
-spec:
- rules:
- - host: test1.apisix.apache.org
- http:
- paths:
- - backend:
- serviceName: api6
- servicePort: 80
- path: /test*
- plugins:
- - config:
- key: apisix-chash-key
- uri_args:
- - pId
- - userId|device
- enable: true
- name: aispeech-chash
- - backend:
- serviceName: httpserver
- servicePort: 8080
- path: /hello*
- plugins:
- - config:
- key: apisix-chash-key
- uri_args:
- - productId2
- - productId|deviceName
- enable: true
- name: aispeech-chash
-`
diff --git a/pkg/ingress/apisix/tls.go b/pkg/ingress/apisix/tls.go
index eb547b6..856adae 100644
--- a/pkg/ingress/apisix/tls.go
+++ b/pkg/ingress/apisix/tls.go
@@ -22,7 +22,6 @@ import (
ingressConf "github.com/apache/apisix-ingress-controller/pkg/kube"
configv1
"github.com/apache/apisix-ingress-controller/pkg/kube/apisix/apis/config/v1"
- "github.com/apache/apisix-ingress-controller/pkg/seven/conf"
apisix "github.com/apache/apisix-ingress-controller/pkg/types/apisix/v1"
)
@@ -32,8 +31,6 @@ type ApisixTLSCRD configv1.ApisixTls
func (as *ApisixTLSCRD) Convert(sc Secreter) (*apisix.Ssl, error) {
name := as.Name
namespace := as.Namespace
- _, group := BuildAnnotation(as.Annotations)
- conf.AddGroup(group)
id := namespace + "_" + name
secretName := as.Spec.Secret.Name
@@ -55,7 +52,6 @@ func (as *ApisixTLSCRD) Convert(sc Secreter) (*apisix.Ssl,
error) {
Cert: cert,
Key: key,
Status: status,
- Group: group,
}
return ssl, nil
}
diff --git a/pkg/ingress/apisix/tls_test.go b/pkg/ingress/apisix/tls_test.go
index 9f67d27..4f5a3a1 100644
--- a/pkg/ingress/apisix/tls_test.go
+++ b/pkg/ingress/apisix/tls_test.go
@@ -69,46 +69,6 @@ spec:
assert.EqualValues(t, sslExpect.Group, ssl.Group, "group convert error")
}
-func TestConvert_group_annotation(t *testing.T) {
- atlsStr := `
-apiVersion: apisix.apache.org/v1
-kind: ApisixTls
-metadata:
- annotations:
- k8s.apisix.apache.org/ingress.class: 127.0.0.1:9080
- name: foo
- namespace: helm
-spec:
- hosts:
- - api6.com
- secret:
- name: test-atls
- namespace: helm
-`
- id := "helm_foo"
- snis := []string{"api6.com"}
- status := int(1)
- cert := "root"
- key := "123456"
- group := "127.0.0.1:9080"
- sslExpect := &apisix.Ssl{
- ID: id,
- Snis: snis,
- Cert: cert,
- Key: key,
- Status: status,
- Group: group,
- }
- setDummyApisixClient(t)
- atlsCRD := &ApisixTLSCRD{}
- err := yaml.Unmarshal([]byte(atlsStr), atlsCRD)
- assert.Nil(t, err, "yaml decode failed")
- sc := &SecretClientMock{}
- ssl, err := atlsCRD.Convert(sc)
- assert.Nil(t, err)
- assert.EqualValues(t, sslExpect.Group, ssl.Group, "group convert error")
-}
-
func TestConvert_Error(t *testing.T) {
atlsStr := `
apiVersion: apisix.apache.org/v1
diff --git a/pkg/ingress/controller/apisix_route.go
b/pkg/ingress/controller/apisix_route.go
index f629d35..f16ddbb 100644
--- a/pkg/ingress/controller/apisix_route.go
+++ b/pkg/ingress/controller/apisix_route.go
@@ -18,6 +18,8 @@ import (
"fmt"
"time"
+ "go.uber.org/zap"
+
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
@@ -26,7 +28,6 @@ import (
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/util/workqueue"
- "github.com/apache/apisix-ingress-controller/pkg/ingress/apisix"
configv1
"github.com/apache/apisix-ingress-controller/pkg/kube/apisix/apis/config/v1"
clientset
"github.com/apache/apisix-ingress-controller/pkg/kube/apisix/client/clientset/versioned"
apisixscheme
"github.com/apache/apisix-ingress-controller/pkg/kube/apisix/client/clientset/versioned/scheme"
@@ -217,10 +218,23 @@ func (c *ApisixRouteController) add(key string) error {
runtime.HandleError(fmt.Errorf("failed to list ApisixRoute %s:
%s", key, err.Error()))
return err
}
- apisixRoute := apisix.ApisixRoute(*apisixIngressRoute)
- routes, services, upstreams, _ :=
apisixRoute.Convert(c.controller.translator)
- comb := state.ApisixCombination{Routes: routes, Services: services,
Upstreams: upstreams}
+ routes, upstreams, err :=
c.controller.translator.TranslateRouteV1(apisixIngressRoute)
+ if err != nil {
+ log.Errorw("failed to translate ApisixRoute/v1",
+ zap.Any("route", apisixIngressRoute),
+ zap.Error(err),
+ )
+ return err
+ }
+ comb := state.ApisixCombination{Routes: routes, Upstreams: upstreams}
_, err = comb.Solver()
+ if err != nil {
+ log.Errorw("failed to push routes and upstreams to APISIX",
+ zap.Any("routes", routes),
+ zap.Any("upstreams", upstreams),
+ zap.Error(err),
+ )
+ }
return err
}
@@ -245,11 +259,8 @@ func (c *ApisixRouteController) sync(rqo *RouteQueueObj)
error {
}
return err // if error occurred, return
}
- oldApisixRoute := apisix.ApisixRoute(*rqo.OldObj)
- oldRoutes, _, _, _ :=
oldApisixRoute.Convert(c.controller.translator)
-
- newApisixRoute := apisix.ApisixRoute(*apisixIngressRoute)
- newRoutes, _, _, _ :=
newApisixRoute.Convert(c.controller.translator)
+ oldRoutes, _, _ :=
c.controller.translator.TranslateRouteV1(rqo.OldObj)
+ newRoutes, _, _ :=
c.controller.translator.TranslateRouteV1(apisixIngressRoute)
rc := &state.RouteCompare{OldRoutes: oldRoutes, NewRoutes:
newRoutes}
return rc.Sync()
@@ -259,13 +270,12 @@ func (c *ApisixRouteController) sync(rqo *RouteQueueObj)
error {
log.Warnf("Route %s has been covered when retry",
rqo.Key)
return nil
}
- apisixRoute := apisix.ApisixRoute(*rqo.OldObj)
- routes, services, upstreams, _ :=
apisixRoute.Convert(c.controller.translator)
+ routes, upstreams, _ :=
c.controller.translator.TranslateRouteV1(rqo.OldObj)
rc := &state.RouteCompare{OldRoutes: routes, NewRoutes: nil}
if err := rc.Sync(); err != nil {
return err
} else {
- comb := state.ApisixCombination{Routes: nil, Services:
services, Upstreams: upstreams}
+ comb := state.ApisixCombination{Routes: nil, Upstreams:
upstreams}
if err := comb.Remove(); err != nil {
return err
}
diff --git a/pkg/ingress/apisix/annotation.go
b/pkg/kube/translation/annotations.go
similarity index 54%
rename from pkg/ingress/apisix/annotation.go
rename to pkg/kube/translation/annotations.go
index 4a239e8..ea399fc 100644
--- a/pkg/ingress/apisix/annotation.go
+++ b/pkg/kube/translation/annotations.go
@@ -12,47 +12,52 @@
// 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
+package translation
import (
- "strconv"
-
seven "github.com/apache/apisix-ingress-controller/pkg/seven/apisix"
apisix "github.com/apache/apisix-ingress-controller/pkg/types/apisix/v1"
)
-// BuildAnnotation return plugins and group
-func BuildAnnotation(annotations map[string]string) (apisix.Plugins, string) {
+const (
+ _whitelist = "k8s.apisix.apache.org/whitelist-source-range"
+ _enableCors = "k8s.apisix.apache.org/enable-cors"
+ _corsAllowOrigin = "k8s.apisix.apache.org/cors-allow-origin"
+ _corsAllowHeaders = "k8s.apisix.apache.org/cors-allow-headers"
+ _corsAllowMethods = "k8s.apisix.apache.org/cors-allow-methods"
+)
+
+type cors struct {
+ enable bool
+ allowOrigin string
+ allowHeaders string
+ allowMethods string
+}
+
+func (t *translator) TranslateAnnotations(annotations map[string]string)
apisix.Plugins {
+ var c cors
plugins := make(apisix.Plugins)
- cors := &CorsYaml{}
- // ingress.class
- group := ""
for k, v := range annotations {
switch {
- case k == SSLREDIRECT:
- if b, err := strconv.ParseBool(v); err == nil && b {
- // todo add ssl-redirect plugin
- }
- case k == WHITELIST:
+ case k == _whitelist:
ipRestriction := seven.BuildIpRestriction(&v, nil)
plugins["ip-restriction"] = ipRestriction
- case k == ENABLE_CORS:
- cors.SetEnable(v)
- case k == CORS_ALLOW_ORIGIN:
- cors.SetOrigin(v)
- case k == CORS_ALLOW_HEADERS:
- cors.SetHeaders(v)
- case k == CORS_ALLOW_METHODS:
- cors.SetMethods(v)
- case k == INGRESS_CLASS:
- group = v
- default:
- // do nothing
+ case k == _enableCors:
+ if v == "true" {
+ c.enable = true
+ }
+ case k == _corsAllowOrigin:
+ c.allowOrigin = v
+ case k == _corsAllowHeaders:
+ c.allowHeaders = v
+ case k == _corsAllowMethods:
+ c.allowMethods = v
}
}
- // build CORS plugin
- if cors.Enable {
- plugins["aispeech-cors"] = cors.Build()
+ if c.enable {
+ maxAge := int64(3600)
+ plugins["aispeech-cors"] = seven.BuildCors(true,
&c.allowOrigin, &c.allowHeaders,
+ &c.allowMethods, &maxAge)
}
- return plugins, group
+ return plugins
}
diff --git a/pkg/kube/translation/apisix_route.go
b/pkg/kube/translation/apisix_route.go
new file mode 100644
index 0000000..a7909e1
--- /dev/null
+++ b/pkg/kube/translation/apisix_route.go
@@ -0,0 +1,86 @@
+// 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 translation
+
+import (
+ "github.com/apache/apisix-ingress-controller/pkg/id"
+ configv1
"github.com/apache/apisix-ingress-controller/pkg/kube/apisix/apis/config/v1"
+ apisixv1
"github.com/apache/apisix-ingress-controller/pkg/types/apisix/v1"
+)
+
+func (t *translator) TranslateRouteV1(ar *configv1.ApisixRoute)
([]*apisixv1.Route, []*apisixv1.Upstream, error) {
+ var (
+ routes []*apisixv1.Route
+ upstreams []*apisixv1.Upstream
+ )
+
+ plugins := t.TranslateAnnotations(ar.Annotations)
+ upstreamMap := make(map[string]*apisixv1.Upstream)
+
+ for _, r := range ar.Spec.Rules {
+ for _, p := range r.Http.Paths {
+ routeName := r.Host + p.Path
+ upstreamName :=
apisixv1.ComposeUpstreamName(ar.Namespace, p.Backend.ServiceName,
int32(p.Backend.ServicePort))
+
+ pluginMap := make(apisixv1.Plugins)
+ // 1.add annotation plugins
+ for k, v := range plugins {
+ pluginMap[k] = v
+ }
+ // 2.add route plugins
+ for _, plugin := range p.Plugins {
+ if !plugin.Enable {
+ continue
+ }
+ if plugin.Config != nil {
+ pluginMap[plugin.Name] = plugin.Config
+ } else if plugin.ConfigSet != nil {
+ pluginMap[plugin.Name] =
plugin.ConfigSet
+ } else {
+ pluginMap[plugin.Name] =
make(map[string]interface{})
+ }
+ }
+
+ route := &apisixv1.Route{
+ Metadata: apisixv1.Metadata{
+ FullName: routeName,
+ ResourceVersion: ar.ResourceVersion,
+ Name: routeName,
+ },
+ Host: r.Host,
+ Path: p.Path,
+ UpstreamName: upstreamName,
+ UpstreamId: id.GenID(upstreamName),
+ Plugins: pluginMap,
+ }
+ routes = append(routes, route)
+
+ if _, ok := upstreamMap[upstreamName]; !ok {
+ ups, err := t.TranslateUpstream(ar.Namespace,
p.Backend.ServiceName, int32(p.Backend.ServicePort))
+ if err != nil {
+ return nil, nil, err
+ }
+ ups.FullName = upstreamName
+ ups.ResourceVersion = ar.ResourceVersion
+ ups.Name = upstreamName
+ upstreamMap[ups.FullName] = ups
+ }
+ }
+ }
+ for _, ups := range upstreamMap {
+ upstreams = append(upstreams, ups)
+ }
+ return routes, upstreams, nil
+}
diff --git a/pkg/kube/translation/translator.go
b/pkg/kube/translation/translator.go
index 7ad4739..da31d20 100644
--- a/pkg/kube/translation/translator.go
+++ b/pkg/kube/translation/translator.go
@@ -47,7 +47,7 @@ type Translator interface {
TranslateUpstreamNodes(*corev1.Endpoints, int32)
([]apisixv1.UpstreamNode, error)
// TranslateUpstreamConfig translates ApisixUpstreamConfig (part of
ApisixUpstream)
// to APISIX Upstream, it doesn't fill the the Upstream metadata and
nodes.
- TranslateUpstreamConfig(config *configv1.ApisixUpstreamConfig)
(*apisixv1.Upstream, error)
+ TranslateUpstreamConfig(*configv1.ApisixUpstreamConfig)
(*apisixv1.Upstream, error)
// TranslateUpstream composes an upstream according to the
// given namespace, name (searching Service/Endpoints) and port
(filtering Endpoints).
// The returned Upstream doesn't have metadata info.
@@ -57,6 +57,9 @@ type Translator interface {
// TranslateIngress composes a couple of APISIX Routes and upstreams
according
// to the given Ingress resource.
TranslateIngress(kube.Ingress) ([]*apisixv1.Route,
[]*apisixv1.Upstream, error)
+ // TranslateRouteV1 translates the configv1.ApisixRoute object into
several Route
+ // and Upstream resources.
+ TranslateRouteV1(*configv1.ApisixRoute) ([]*apisixv1.Route,
[]*apisixv1.Upstream, error)
}
// TranslatorOptions contains options to help Translator
diff --git a/pkg/seven/state/builder.go b/pkg/seven/state/builder.go
index 82e005f..0e229d3 100644
--- a/pkg/seven/state/builder.go
+++ b/pkg/seven/state/builder.go
@@ -75,13 +75,13 @@ func NewRouteWorkers(ctx context.Context,
for _, r := range routes {
rw := &routeWorker{Route: r, Ctx: ctx, Wg: wg, ErrorChan:
errorChan}
rw.start()
- rwg.Add(r.ServiceName, rw)
+ rwg.Add(r.UpstreamName, rw)
}
return rwg
}
// 3.route get the Event and trigger a padding for object,then diff,sync;
-func (r *routeWorker) trigger(event Event) {
+func (r *routeWorker) trigger() {
var (
op string
errNotify error
@@ -92,10 +92,6 @@ func (r *routeWorker) trigger(event Event) {
}
r.Wg.Done()
}()
- // consumer Event
- service := event.Obj.(*v1.Service)
- r.ServiceId = service.ID
- log.Infof("trigger routeWorker %s from %s, %s", r.Name, event.Op,
service.Name)
// padding
var cluster string
@@ -152,27 +148,14 @@ func (r *routeWorker) sync(op string) error {
return nil
}
-// service
-func NewServiceWorkers(ctx context.Context,
- services []*v1.Service, rwg *RouteWorkerGroup, wg *sync.WaitGroup,
errorChan chan CRDStatus) ServiceWorkerGroup {
- swg := make(ServiceWorkerGroup)
- for _, s := range services {
- rw := &serviceWorker{Service: s, Ctx: ctx, Wg: wg, ErrorChan:
errorChan}
- //rw.Wg.Add(1)
- rw.start(rwg)
- swg.Add(s.UpstreamName, rw)
- }
- return swg
-}
-
// upstream
-func SolverUpstream(upstreams []*v1.Upstream, swg ServiceWorkerGroup, wg
*sync.WaitGroup, errorChan chan CRDStatus) {
+func SolverUpstream(upstreams []*v1.Upstream, rwg RouteWorkerGroup, wg
*sync.WaitGroup, errorChan chan CRDStatus) {
for _, u := range upstreams {
- go SolverSingleUpstream(u, swg, wg, errorChan)
+ go SolverSingleUpstream(u, rwg, wg, errorChan)
}
}
-func SolverSingleUpstream(u *v1.Upstream, swg ServiceWorkerGroup, wg
*sync.WaitGroup, errorChan chan CRDStatus) {
+func SolverSingleUpstream(u *v1.Upstream, rwg RouteWorkerGroup, wg
*sync.WaitGroup, errorChan chan CRDStatus) {
var (
op string
errNotify error
@@ -241,10 +224,8 @@ func SolverSingleUpstream(u *v1.Upstream, swg
ServiceWorkerGroup, wg *sync.WaitG
}
}
log.Infof("solver upstream %s:%s", op, u.Name)
- // anyway, broadcast to service
- serviceWorkers := swg[u.Name]
- for _, sw := range serviceWorkers {
- event := &Event{Kind: UpstreamKind, Op: op, Obj: u}
- sw.Event <- *event
+ // anyway, broadcast to route
+ for _, sw := range rwg[u.Name] {
+ sw.Event <- Event{}
}
}
diff --git a/pkg/seven/state/route_worker.go b/pkg/seven/state/route_worker.go
index 30054fc..ed28234 100644
--- a/pkg/seven/state/route_worker.go
+++ b/pkg/seven/state/route_worker.go
@@ -38,8 +38,8 @@ func (w *routeWorker) start() {
go func() {
for {
select {
- case event := <-w.Event:
- w.trigger(event)
+ case <-w.Event:
+ w.trigger()
case <-w.Ctx.Done():
return
}
diff --git a/pkg/seven/state/service_worker.go
b/pkg/seven/state/service_worker.go
deleted file mode 100644
index d57d781..0000000
--- a/pkg/seven/state/service_worker.go
+++ /dev/null
@@ -1,147 +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 state
-
-import (
- "context"
- "sync"
-
- "github.com/apache/apisix-ingress-controller/pkg/log"
- "github.com/apache/apisix-ingress-controller/pkg/seven/conf"
- "github.com/apache/apisix-ingress-controller/pkg/seven/utils"
- v1 "github.com/apache/apisix-ingress-controller/pkg/types/apisix/v1"
-)
-
-type serviceWorker struct {
- *v1.Service
- Event chan Event
- Ctx context.Context
- Wg *sync.WaitGroup
- ErrorChan chan CRDStatus
-}
-
-// ServiceWorkerGroup for broadcast from upstream to service
-type ServiceWorkerGroup map[string][]*serviceWorker
-
-// start start watch event
-func (w *serviceWorker) start(rwg *RouteWorkerGroup) {
- w.Event = make(chan Event)
- go func() {
- for {
- select {
- case event := <-w.Event:
- if err := w.trigger(event, rwg); err != nil {
- log.Errorf("failed to trigger event:
%s", err)
- }
- case <-w.Ctx.Done():
- return
- }
- }
- }()
-}
-
-// trigger add to queue
-func (w *serviceWorker) trigger(event Event, rwg *RouteWorkerGroup) error {
- log.Infof("1.service trigger from %s, %s", event.Op, event.Kind)
- // consumer Event set upstreamID
- upstream := event.Obj.(*v1.Upstream)
- log.Infof("2.service trigger from %s, %s", event.Op, upstream.Name)
-
- w.UpstreamId = upstream.ID
- // add to queue
- services := []*v1.Service{w.Service}
- sqo := &ServiceQueueObj{Services: services, RouteWorkerGroup: *rwg}
- //sqo.AddQueue()
-
- SolverService(sqo.Services, sqo.RouteWorkerGroup, w.Wg, w.ErrorChan)
- return nil
-}
-
-func SolverService(services []*v1.Service, rwg RouteWorkerGroup, wg
*sync.WaitGroup, errorChan chan CRDStatus) {
- for _, svc := range services {
- go SolverSingleService(svc, rwg, wg, errorChan)
- }
-}
-
-func SolverSingleService(svc *v1.Service, rwg RouteWorkerGroup, wg
*sync.WaitGroup, errorChan chan CRDStatus) {
- var errNotify error
- defer func() {
- if errNotify != nil {
- errorChan <- CRDStatus{Id: "", Status: "failure", Err:
errNotify}
- }
- wg.Done()
- }()
-
- op := Update
- // padding
- var cluster string
- if svc.Group != "" {
- cluster = svc.Group
- }
- currentService, _ :=
conf.Client.Cluster(cluster).Service().Get(context.TODO(), svc.FullName)
- if paddingService(svc, currentService) {
- op = Create
- }
- // diff
- hasDiff, err := utils.HasDiff(svc, currentService)
- // sync
- if err != nil {
- errNotify = err
- return
- }
- if hasDiff {
- if op == Create {
- if _, err :=
conf.Client.Cluster(cluster).Service().Create(context.TODO(), svc); err != nil {
- log.Errorf("failed to create service: %s", err)
- errNotify = err
- return
- }
- log.Infof("create service %s, %s", svc.Name,
svc.UpstreamId)
- } else {
- if _, err :=
conf.Client.Cluster(cluster).Service().Update(context.TODO(), svc); err != nil {
- errNotify = err
- log.Errorf("failed to update service: %s,
id:%s", err, svc.ID)
- } else {
- log.Infof("updated service, id:%s,
upstream_id:%s", svc.ID, svc.UpstreamId)
- }
- }
- }
- // broadcast to route
- for _, rw := range rwg[svc.Name] {
- event := &Event{Kind: ServiceKind, Op: op, Obj: svc}
- log.Infof("send event %s, %s, %s", event.Kind, event.Op,
svc.Name)
- rw.Event <- *event
- }
-}
-
-func (swg *ServiceWorkerGroup) Add(key string, s *serviceWorker) {
- sws := (*swg)[key]
- if sws == nil {
- sws = make([]*serviceWorker, 0)
- }
- sws = append(sws, s)
- (*swg)[key] = sws
-}
-
-func (swg *ServiceWorkerGroup) Delete(key string, s *serviceWorker) {
- sws := (*swg)[key]
- result := make([]*serviceWorker, 0)
- for _, r := range sws {
- if r.Name != s.Name {
- result = append(result, r)
- }
- }
- (*swg)[key] = result
-}
diff --git a/pkg/seven/state/solver.go b/pkg/seven/state/solver.go
index 48a7a92..838cee7 100644
--- a/pkg/seven/state/solver.go
+++ b/pkg/seven/state/solver.go
@@ -38,7 +38,7 @@ func init() {
func WatchUpstream() {
for {
uqo := <-UpstreamQueue
- SolverUpstream(uqo.Upstreams, uqo.ServiceWorkerGroup, uqo.Wg,
uqo.ErrorChan)
+ SolverUpstream(uqo.Upstreams, uqo.RouteWorkerGroup, uqo.Wg,
uqo.ErrorChan)
}
}
@@ -132,10 +132,8 @@ func (s *ApisixCombination) SyncWithGroup(ctx
context.Context, id string, result
// goroutine for sync route/service/upstream
// route
rwg := NewRouteWorkers(ctx, s.Routes, &wg, resultChan)
- // service
- swg := NewServiceWorkers(ctx, s.Services, &rwg, &wg, resultChan)
// upstream
- uqo := &UpstreamQueueObj{Upstreams: s.Upstreams, ServiceWorkerGroup:
swg, Wg: &wg, ErrorChan: resultChan}
+ uqo := &UpstreamQueueObj{Upstreams: s.Upstreams, RouteWorkerGroup: rwg,
Wg: &wg, ErrorChan: resultChan}
uqo.AddQueue()
waitTimeout(ctx, &wg, resultChan)
@@ -148,10 +146,10 @@ func WaitWorkerGroup(id string, resultChan chan
CRDStatus) (string, error) {
// UpstreamQueueObj for upstream queue
type UpstreamQueueObj struct {
- Upstreams []*v1.Upstream
- ServiceWorkerGroup ServiceWorkerGroup
- Wg *sync.WaitGroup
- ErrorChan chan CRDStatus
+ Upstreams []*v1.Upstream
+ RouteWorkerGroup RouteWorkerGroup
+ Wg *sync.WaitGroup
+ ErrorChan chan CRDStatus
}
type CRDStatus struct {
