You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@apisix.apache.org by to...@apache.org on 2021/01/20 08:52:14 UTC
[apisix-ingress-controller] branch master updated: chore: caching
(#176)
This is an automated email from the ASF dual-hosted git repository.
tokers 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 cc15856 chore: caching (#176)
cc15856 is described below
commit cc158567ddf2c9934aa922db74d4d469ac02dad9
Author: Alex Zhang <zc...@gmail.com>
AuthorDate: Wed Jan 20 16:52:06 2021 +0800
chore: caching (#176)
---
cmd/ingress/ingress.go | 2 +-
pkg/apisix/apisix.go | 31 ++--
pkg/apisix/cache/cache.go | 61 ++++++++
pkg/apisix/cache/memdb.go | 251 +++++++++++++++++++++++++++++++
pkg/apisix/cache/memdb_test.go | 266 +++++++++++++++++++++++++++++++++
pkg/apisix/cache/schema.go | 96 ++++++++++++
pkg/apisix/cluster.go | 145 ++++++++++++++++--
pkg/apisix/cluster_test.go | 10 +-
pkg/apisix/nonexistentclient.go | 46 ++++++
pkg/apisix/route.go | 104 ++++++++++++-
pkg/apisix/route_test.go | 11 +-
pkg/apisix/service.go | 105 ++++++++++++-
pkg/apisix/service_test.go | 8 +-
pkg/apisix/ssl.go | 75 +++++++++-
pkg/apisix/ssl_test.go | 8 +-
pkg/apisix/upstream.go | 106 +++++++++++--
pkg/apisix/upstream_test.go | 8 +-
pkg/ingress/apisix/tls_test.go | 4 +-
pkg/ingress/controller/apisix_route.go | 3 +-
pkg/ingress/controller/controller.go | 25 +++-
pkg/seven/apisix/client.go | 42 ------
pkg/seven/apisix/error.go | 22 ---
pkg/seven/apisix/route.go | 55 -------
pkg/seven/apisix/service.go | 53 -------
pkg/seven/apisix/upstream.go | 70 ---------
pkg/seven/db/route.go | 130 ----------------
pkg/seven/db/service.go | 130 ----------------
pkg/seven/db/store.go | 43 ------
pkg/seven/db/upstream.go | 116 --------------
pkg/seven/state/builder.go | 126 ++++++----------
pkg/seven/state/service_worker.go | 27 +---
pkg/seven/state/solver.go | 156 ++++++++-----------
pkg/seven/utils/diff.go | 1 +
test/e2e/scaffold/crd.go | 6 +-
34 files changed, 1399 insertions(+), 943 deletions(-)
diff --git a/cmd/ingress/ingress.go b/cmd/ingress/ingress.go
index dcaf894..d1dab25 100644
--- a/cmd/ingress/ingress.go
+++ b/cmd/ingress/ingress.go
@@ -109,7 +109,7 @@ the apisix cluster and others are created`,
}
go func() {
if err := ingress.Run(stop); err != nil {
- dief("failed to launch ingress controller: %s", err)
+ dief("failed to run ingress controller: %s", err)
}
}()
diff --git a/pkg/apisix/apisix.go b/pkg/apisix/apisix.go
index 5069f52..1437505 100644
--- a/pkg/apisix/apisix.go
+++ b/pkg/apisix/apisix.go
@@ -41,11 +41,17 @@ type Cluster interface {
Service() Service
// SSL returns a SSL interface that can operate SSL resources.
SSL() SSL
+ // String exposes the client information in human readable format.
+ String() string
+ // Ready waits until all resources in APISIX cluster is synced to
+ // cache.
+ HasSynced(context.Context) error
}
// Route is the specific client interface to take over the create, update,
// list and delete for APISIX's Route resource.
type Route interface {
+ Get(context.Context, string) (*v1.Route, error)
List(context.Context) ([]*v1.Route, error)
Create(context.Context, *v1.Route) (*v1.Route, error)
Delete(context.Context, *v1.Route) error
@@ -55,6 +61,7 @@ type Route interface {
// SSL is the specific client interface to take over the create, update,
// list and delete for APISIX's SSL resource.
type SSL interface {
+ Get(context.Context, string) (*v1.Ssl, error)
List(context.Context) ([]*v1.Ssl, error)
Create(context.Context, *v1.Ssl) (*v1.Ssl, error)
Delete(context.Context, *v1.Ssl) error
@@ -64,6 +71,7 @@ type SSL interface {
// Upstream is the specific client interface to take over the create, update,
// list and delete for APISIX's Upstream resource.
type Upstream interface {
+ Get(context.Context, string) (*v1.Upstream, error)
List(context.Context) ([]*v1.Upstream, error)
Create(context.Context, *v1.Upstream) (*v1.Upstream, error)
Delete(context.Context, *v1.Upstream) error
@@ -73,6 +81,7 @@ type Upstream interface {
// Service is the specific client interface to take over the create, update,
// list and delete for APISIX's Service resource.
type Service interface {
+ Get(context.Context, string) (*v1.Service, error)
List(context.Context) ([]*v1.Service, error)
Create(context.Context, *v1.Service) (*v1.Service, error)
Delete(context.Context, *v1.Service) error
@@ -80,22 +89,13 @@ type Service interface {
}
type apisix struct {
- defaultCluster Cluster
nonExistentCluster Cluster
- defaultClusterName string
clusters map[string]Cluster
}
-// NewForOptions creates an APISIX client to perform resources change pushing.
-// Users should carry a ClusterOptions to configure the default APISIX cluster.
-func NewForOptions(co *ClusterOptions) (APISIX, error) {
- defaultCluster, err := newCluster(co)
- if err != nil {
- return nil, err
- }
+// NewClient creates an APISIX client to perform resources change pushing.
+func NewClient() (APISIX, error) {
cli := &apisix{
- defaultCluster: defaultCluster,
- defaultClusterName: co.Name,
nonExistentCluster: newNonExistentCluster(),
}
return cli, nil
@@ -103,9 +103,6 @@ func NewForOptions(co *ClusterOptions) (APISIX, error) {
// Cluster implements APISIX.Cluster method.
func (c *apisix) Cluster(name string) Cluster {
- if name == c.defaultClusterName {
- return c.defaultCluster
- }
cluster, ok := c.clusters[name]
if !ok {
return c.nonExistentCluster
@@ -115,8 +112,7 @@ func (c *apisix) Cluster(name string) Cluster {
// ListClusters implements APISIX.ListClusters method.
func (c *apisix) ListClusters() []Cluster {
- clusters := make([]Cluster, 0, len(c.clusters)+1)
- clusters = append(clusters, c.defaultCluster)
+ clusters := make([]Cluster, 0, len(c.clusters))
for _, cluster := range c.clusters {
clusters = append(clusters, cluster)
}
@@ -125,9 +121,6 @@ func (c *apisix) ListClusters() []Cluster {
// AddCluster implements APISIX.AddCluster method.
func (c *apisix) AddCluster(co *ClusterOptions) error {
- if co.Name == c.defaultClusterName {
- return ErrDuplicatedCluster
- }
_, ok := c.clusters[co.Name]
if ok {
return ErrDuplicatedCluster
diff --git a/pkg/apisix/cache/cache.go b/pkg/apisix/cache/cache.go
new file mode 100644
index 0000000..6e17e26
--- /dev/null
+++ b/pkg/apisix/cache/cache.go
@@ -0,0 +1,61 @@
+// 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 cache
+
+import v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
+
+// Cache defines the necessary behaviors that the cache object should have.
+// Note this interface is for APISIX, not for generic purpose, it supports
+// standard APISIX resources, i.e. Route, Upstream, Service and SSL.
+// Cache implementations should copy the target objects before/after read/write
+// operations for the sake of avoiding data corrupted by other writers.
+type Cache interface {
+ // InsertRoute adds or updates route to cache.
+ InsertRoute(*v1.Route) error
+ // InsertService adds or updates service to cache.
+ InsertService(*v1.Service) error
+ // InsertSSL adds or updates ssl to cache.
+ InsertSSL(*v1.Ssl) error
+ // InsertUpstream adds or updates upstream to cache.
+ InsertUpstream(*v1.Upstream) error
+
+ // GetRoute finds the route from cache according to the primary index.
+ GetRoute(string) (*v1.Route, error)
+ // GetService finds the service from cache according to the primary index.
+ GetService(string) (*v1.Service, error)
+ // GetSSL finds the ssl from cache according to the primary index.
+ GetSSL(string) (*v1.Ssl, error)
+ // GetUpstream finds the upstream from cache according to the primary index.
+ GetUpstream(string) (*v1.Upstream, error)
+
+ // ListRoutes lists all routes in cache.
+ ListRoutes() ([]*v1.Route, error)
+ // ListServices lists all services in cache.
+ ListServices() ([]*v1.Service, error)
+ // ListSSL lists all ssl objects in cache.
+ ListSSL() ([]*v1.Ssl, error)
+ // ListUpstreams lists all upstreams in cache.
+ ListUpstreams() ([]*v1.Upstream, error)
+
+ // DeleteRoute deletes the specified route in cache.
+ DeleteRoute(*v1.Route) error
+ // DeleteService deletes the specified service in cache.
+ DeleteService(*v1.Service) error
+ // DeleteSSL deletes the specified ssl in cache.
+ DeleteSSL(*v1.Ssl) error
+ // DeleteUpstream deletes the specified upstream in cache.
+ DeleteUpstream(*v1.Upstream) error
+}
diff --git a/pkg/apisix/cache/memdb.go b/pkg/apisix/cache/memdb.go
new file mode 100644
index 0000000..aa7e8f6
--- /dev/null
+++ b/pkg/apisix/cache/memdb.go
@@ -0,0 +1,251 @@
+// 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 cache
+
+import (
+ "errors"
+
+ "github.com/hashicorp/go-memdb"
+
+ v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
+)
+
+var (
+ // ErrStillInUse means an object is still in use.
+ ErrStillInUse = errors.New("still in use")
+ // ErrNotFound is returned when the requested item is not found.
+ ErrNotFound = memdb.ErrNotFound
+)
+
+type dbCache struct {
+ db *memdb.MemDB
+}
+
+// NewMemDBCache creates a Cache object backs with a memory DB.
+func NewMemDBCache() (Cache, error) {
+ db, err := memdb.NewMemDB(_schema)
+ if err != nil {
+ return nil, err
+ }
+ return &dbCache{
+ db: db,
+ }, nil
+}
+
+func (c *dbCache) InsertRoute(r *v1.Route) error {
+ return c.insert("route", r.DeepCopy())
+}
+
+func (c *dbCache) InsertService(s *v1.Service) error {
+ return c.insert("service", s.DeepCopy())
+}
+
+func (c *dbCache) InsertSSL(ssl *v1.Ssl) error {
+ return c.insert("ssl", ssl.DeepCopy())
+}
+
+func (c *dbCache) InsertUpstream(u *v1.Upstream) error {
+ return c.insert("upstream", u.DeepCopy())
+}
+
+func (c *dbCache) insert(table string, obj interface{}) error {
+ txn := c.db.Txn(true)
+ defer txn.Abort()
+ if err := txn.Insert(table, obj); err != nil {
+ return err
+ }
+ txn.Commit()
+ return nil
+}
+
+func (c *dbCache) GetRoute(key string) (*v1.Route, error) {
+ obj, err := c.get("route", key)
+ if err != nil {
+ return nil, err
+ }
+ return obj.(*v1.Route).DeepCopy(), nil
+}
+
+func (c *dbCache) GetService(key string) (*v1.Service, error) {
+ obj, err := c.get("service", key)
+ if err != nil {
+ return nil, err
+ }
+ return obj.(*v1.Service).DeepCopy(), nil
+}
+
+func (c *dbCache) GetSSL(key string) (*v1.Ssl, error) {
+ obj, err := c.get("ssl", key)
+ if err != nil {
+ return nil, err
+ }
+ return obj.(*v1.Ssl).DeepCopy(), nil
+}
+
+func (c *dbCache) GetUpstream(key string) (*v1.Upstream, error) {
+ obj, err := c.get("upstream", key)
+ if err != nil {
+ return nil, err
+ }
+ return obj.(*v1.Upstream).DeepCopy(), nil
+}
+
+func (c *dbCache) get(table, key string) (interface{}, error) {
+ txn := c.db.Txn(false)
+ defer txn.Abort()
+ obj, err := txn.First(table, "id", key)
+ if err != nil {
+ if err == memdb.ErrNotFound {
+ return nil, ErrNotFound
+ }
+ return nil, err
+ }
+ if obj == nil {
+ return nil, ErrNotFound
+ }
+ return obj, nil
+}
+
+func (c *dbCache) ListRoutes() ([]*v1.Route, error) {
+ raws, err := c.list("route")
+ if err != nil {
+ return nil, err
+ }
+ routes := make([]*v1.Route, 0, len(raws))
+ for _, raw := range raws {
+ routes = append(routes, raw.(*v1.Route).DeepCopy())
+ }
+ return routes, nil
+}
+
+func (c *dbCache) ListServices() ([]*v1.Service, error) {
+ raws, err := c.list("service")
+ if err != nil {
+ return nil, err
+ }
+ services := make([]*v1.Service, 0, len(raws))
+ for _, raw := range raws {
+ services = append(services, raw.(*v1.Service).DeepCopy())
+ }
+ return services, nil
+}
+
+func (c *dbCache) ListSSL() ([]*v1.Ssl, error) {
+ raws, err := c.list("ssl")
+ if err != nil {
+ return nil, err
+ }
+ ssl := make([]*v1.Ssl, 0, len(raws))
+ for _, raw := range raws {
+ ssl = append(ssl, raw.(*v1.Ssl).DeepCopy())
+ }
+ return ssl, nil
+}
+
+func (c *dbCache) ListUpstreams() ([]*v1.Upstream, error) {
+ raws, err := c.list("upstream")
+ if err != nil {
+ return nil, err
+ }
+ upstreams := make([]*v1.Upstream, 0, len(raws))
+ for _, raw := range raws {
+ upstreams = append(upstreams, raw.(*v1.Upstream).DeepCopy())
+ }
+ return upstreams, nil
+}
+
+func (c *dbCache) list(table string) ([]interface{}, error) {
+ txn := c.db.Txn(false)
+ defer txn.Abort()
+ iter, err := txn.Get(table, "id")
+ if err != nil {
+ return nil, err
+ }
+ var objs []interface{}
+ for obj := iter.Next(); obj != nil; obj = iter.Next() {
+ objs = append(objs, obj)
+ }
+ return objs, nil
+}
+
+func (c *dbCache) DeleteRoute(r *v1.Route) error {
+ return c.delete("route", r)
+}
+
+func (c *dbCache) DeleteService(s *v1.Service) error {
+ if err := c.checkServiceReference(s); err != nil {
+ return err
+ }
+ return c.delete("service", s)
+}
+
+func (c *dbCache) DeleteSSL(ssl *v1.Ssl) error {
+ return c.delete("ssl", ssl)
+}
+
+func (c *dbCache) DeleteUpstream(u *v1.Upstream) error {
+ if err := c.checkUpstreamReference(u); err != nil {
+ return err
+ }
+ return c.delete("upstream", u)
+}
+
+func (c *dbCache) delete(table string, obj interface{}) error {
+ txn := c.db.Txn(true)
+ defer txn.Abort()
+ if err := txn.Delete(table, obj); err != nil {
+ if err == memdb.ErrNotFound {
+ return ErrNotFound
+ }
+ return err
+ }
+ txn.Commit()
+ return nil
+}
+
+func (c *dbCache) checkServiceReference(s *v1.Service) error {
+ // Service is referenced by Route.
+ txn := c.db.Txn(false)
+ defer txn.Abort()
+ obj, err := txn.First("route", "service_id", *s.FullName)
+ if err != nil {
+ if err == memdb.ErrNotFound {
+ return nil
+ }
+ return err
+ }
+ if obj == nil {
+ return nil
+ }
+ return ErrStillInUse
+}
+
+func (c *dbCache) checkUpstreamReference(u *v1.Upstream) error {
+ // Upstream is referenced by Service.
+ txn := c.db.Txn(false)
+ defer txn.Abort()
+ obj, err := txn.First("service", "upstream_id", *u.FullName)
+ if err != nil {
+ if err == memdb.ErrNotFound {
+ return nil
+ }
+ return err
+ }
+ if obj == nil {
+ return nil
+ }
+ return ErrStillInUse
+}
diff --git a/pkg/apisix/cache/memdb_test.go b/pkg/apisix/cache/memdb_test.go
new file mode 100644
index 0000000..e16bffe
--- /dev/null
+++ b/pkg/apisix/cache/memdb_test.go
@@ -0,0 +1,266 @@
+// 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 cache
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+
+ v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
+)
+
+func TestMemDBCacheRoute(t *testing.T) {
+ c, err := NewMemDBCache()
+ assert.Nil(t, err, "NewMemDBCache")
+
+ name := "abc"
+ sid := "1"
+ r1 := &v1.Route{
+ FullName: &name,
+ Name: &name,
+ ServiceId: &sid,
+ }
+ assert.Nil(t, c.InsertRoute(r1), "inserting route 1")
+
+ r, err := c.GetRoute("abc")
+ assert.Equal(t, r1, r)
+
+ name2 := "def"
+ sid2 := "2"
+ r2 := &v1.Route{
+ FullName: &name2,
+ Name: &name2,
+ ServiceId: &sid2,
+ }
+ name3 := "ghi"
+ sid3 := "3"
+ r3 := &v1.Route{
+ FullName: &name3,
+ Name: &name3,
+ ServiceId: &sid3,
+ }
+ assert.Nil(t, c.InsertRoute(r2), "inserting route r2")
+ assert.Nil(t, c.InsertRoute(r3), "inserting route r3")
+
+ r, err = c.GetRoute("ghi")
+ assert.Equal(t, r3, r)
+
+ assert.Nil(t, c.DeleteRoute(r3), "delete route r3")
+
+ routes, err := c.ListRoutes()
+ assert.Nil(t, err, "listing routes")
+
+ if *routes[0].FullName > *routes[1].FullName {
+ routes[0], routes[1] = routes[1], routes[0]
+ }
+ assert.Equal(t, routes[0], r1)
+ assert.Equal(t, routes[1], r2)
+
+ name4 := "name4"
+ sid4 := "4"
+ r4 := &v1.Route{
+ FullName: &name4,
+ Name: &name4,
+ ServiceId: &sid4,
+ }
+ assert.Error(t, ErrNotFound, c.DeleteRoute(r4))
+}
+
+func TestMemDBCacheService(t *testing.T) {
+ c, err := NewMemDBCache()
+ assert.Nil(t, err, "NewMemDBCache")
+
+ name := "abc"
+ uid := "1"
+ s1 := &v1.Service{
+ FullName: &name,
+ Name: &name,
+ UpstreamId: &uid,
+ }
+ assert.Nil(t, c.InsertService(s1), "inserting service 1")
+
+ s, err := c.GetService("abc")
+ assert.Equal(t, s1, s)
+
+ name2 := "def"
+ uid2 := "2"
+ s2 := &v1.Service{
+ FullName: &name2,
+ Name: &name2,
+ UpstreamId: &uid2,
+ }
+ name3 := "ghi"
+ uid3 := "3"
+ s3 := &v1.Service{
+ FullName: &name3,
+ Name: &name3,
+ UpstreamId: &uid3,
+ }
+ assert.Nil(t, c.InsertService(s2), "inserting service 2")
+ assert.Nil(t, c.InsertService(s3), "inserting service 3")
+
+ s, err = c.GetService("ghi")
+ assert.Equal(t, s3, s)
+
+ assert.Nil(t, c.DeleteService(s3), "delete service 3")
+
+ services, err := c.ListServices()
+ assert.Nil(t, err, "listing services")
+
+ if *services[0].FullName > *services[1].FullName {
+ services[0], services[1] = services[1], services[0]
+ }
+ assert.Equal(t, services[0], s1)
+ assert.Equal(t, services[1], s2)
+
+ name4 := "name4"
+ uid4 := "4"
+ s4 := &v1.Service{
+ FullName: &name4,
+ Name: &name4,
+ UpstreamId: &uid4,
+ }
+ assert.Error(t, ErrNotFound, c.DeleteService(s4))
+}
+
+func TestMemDBCacheSSL(t *testing.T) {
+ c, err := NewMemDBCache()
+ assert.Nil(t, err, "NewMemDBCache")
+
+ id := "abc"
+ s1 := &v1.Ssl{
+ ID: &id,
+ }
+ assert.Nil(t, c.InsertSSL(s1), "inserting ssl 1")
+
+ s, err := c.GetSSL("abc")
+ assert.Equal(t, s1, s)
+
+ id2 := "def"
+ s2 := &v1.Ssl{
+ ID: &id2,
+ }
+ id3 := "ghi"
+ s3 := &v1.Ssl{
+ ID: &id3,
+ }
+ assert.Nil(t, c.InsertSSL(s2), "inserting ssl 2")
+ assert.Nil(t, c.InsertSSL(s3), "inserting ssl 3")
+
+ s, err = c.GetSSL("ghi")
+ assert.Equal(t, s3, s)
+
+ assert.Nil(t, c.DeleteSSL(s3), "delete ssl 3")
+
+ ssl, err := c.ListSSL()
+ assert.Nil(t, err, "listing ssl")
+
+ if *ssl[0].ID > *ssl[1].ID {
+ ssl[0], ssl[1] = ssl[1], ssl[0]
+ }
+ assert.Equal(t, ssl[0], s1)
+ assert.Equal(t, ssl[1], s2)
+
+ id4 := "id4"
+ s4 := &v1.Ssl{
+ ID: &id4,
+ }
+ assert.Error(t, ErrNotFound, c.DeleteSSL(s4))
+}
+
+func TestMemDBCacheUpstream(t *testing.T) {
+ c, err := NewMemDBCache()
+ assert.Nil(t, err, "NewMemDBCache")
+
+ name := "abc"
+ u1 := &v1.Upstream{
+ FullName: &name,
+ Name: &name,
+ }
+ assert.Nil(t, c.InsertUpstream(u1), "inserting upstream 1")
+
+ u, err := c.GetUpstream("abc")
+ assert.Equal(t, u1, u)
+
+ name2 := "def"
+ u2 := &v1.Upstream{
+ FullName: &name2,
+ Name: &name2,
+ }
+ name3 := "ghi"
+ u3 := &v1.Upstream{
+ FullName: &name3,
+ Name: &name3,
+ }
+ assert.Nil(t, c.InsertUpstream(u2), "inserting upstream 2")
+ assert.Nil(t, c.InsertUpstream(u3), "inserting upstream 3")
+
+ u, err = c.GetUpstream("ghi")
+ assert.Equal(t, u3, u)
+
+ assert.Nil(t, c.DeleteUpstream(u3), "delete upstream 3")
+
+ upstreams, err := c.ListUpstreams()
+ assert.Nil(t, err, "listing upstreams")
+
+ if *upstreams[0].FullName > *upstreams[1].FullName {
+ upstreams[0], upstreams[1] = upstreams[1], upstreams[0]
+ }
+ assert.Equal(t, upstreams[0], u1)
+ assert.Equal(t, upstreams[1], u2)
+
+ name4 := "name4"
+ u4 := &v1.Upstream{
+ FullName: &name4,
+ Name: &name4,
+ }
+ assert.Error(t, ErrNotFound, c.DeleteUpstream(u4))
+}
+
+func TestMemDBCacheReference(t *testing.T) {
+ rname := "route"
+ sid := "service"
+ r := &v1.Route{
+ FullName: &rname,
+ Name: &rname,
+ ServiceId: &sid,
+ }
+ sname := "service"
+ uid := "upstream"
+ s := &v1.Service{
+ FullName: &sname,
+ Name: &sname,
+ UpstreamId: &uid,
+ }
+ uname := "upstream"
+ u := &v1.Upstream{
+ FullName: &uname,
+ Name: &uname,
+ }
+
+ db, err := NewMemDBCache()
+ assert.Nil(t, err, "NewMemDBCache")
+ assert.Nil(t, db.InsertRoute(r))
+ assert.Nil(t, db.InsertService(s))
+ assert.Nil(t, db.InsertUpstream(u))
+
+ assert.Error(t, ErrStillInUse, db.DeleteService(s))
+ assert.Error(t, ErrStillInUse, db.DeleteUpstream(u))
+ assert.Nil(t, db.DeleteRoute(r))
+ assert.Nil(t, db.DeleteService(s))
+ assert.Nil(t, db.DeleteUpstream(u))
+}
diff --git a/pkg/apisix/cache/schema.go b/pkg/apisix/cache/schema.go
new file mode 100644
index 0000000..f7fa6f0
--- /dev/null
+++ b/pkg/apisix/cache/schema.go
@@ -0,0 +1,96 @@
+// 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 cache
+
+import (
+ "github.com/hashicorp/go-memdb"
+)
+
+var (
+ _schema = &memdb.DBSchema{
+ Tables: map[string]*memdb.TableSchema{
+ "route": {
+ Name: "route",
+ Indexes: map[string]*memdb.IndexSchema{
+ "id": {
+ Name: "id",
+ Unique: true,
+ Indexer: &memdb.StringFieldIndex{Field: "FullName"},
+ },
+ "name": {
+ Name: "name",
+ Unique: true,
+ Indexer: &memdb.StringFieldIndex{Field: "Name"},
+ AllowMissing: true,
+ },
+ "service_id": {
+ Name: "service_id",
+ Unique: false,
+ Indexer: &memdb.StringFieldIndex{Field: "ServiceId"},
+ },
+ },
+ },
+ "service": {
+ Name: "service",
+ Indexes: map[string]*memdb.IndexSchema{
+ "id": {
+ Name: "id",
+ Unique: true,
+ Indexer: &memdb.StringFieldIndex{Field: "FullName"},
+ },
+ "name": {
+ Name: "name",
+ Unique: true,
+ Indexer: &memdb.StringFieldIndex{Field: "Name"},
+ AllowMissing: true,
+ },
+ "upstream_id": {
+ Name: "upstream_id",
+ Unique: false,
+ Indexer: &memdb.StringFieldIndex{Field: "UpstreamId"},
+ AllowMissing: true,
+ },
+ },
+ },
+ "upstream": {
+ Name: "upstream",
+ Indexes: map[string]*memdb.IndexSchema{
+ "id": {
+ Name: "id",
+ Unique: true,
+ Indexer: &memdb.StringFieldIndex{Field: "FullName"},
+ },
+ "name": {
+ Name: "name",
+ Unique: true,
+ Indexer: &memdb.StringFieldIndex{Field: "Name"},
+ AllowMissing: true,
+ },
+ },
+ },
+ "ssl": {
+ Name: "ssl",
+ Indexes: map[string]*memdb.IndexSchema{
+ "id": {
+ Name: "id",
+ Unique: true,
+ Indexer: &memdb.StringFieldIndex{Field: "ID"},
+ },
+ },
+ },
+ },
+ }
+)
diff --git a/pkg/apisix/cluster.go b/pkg/apisix/cluster.go
index 0a9cab7..4a52412 100644
--- a/pkg/apisix/cluster.go
+++ b/pkg/apisix/cluster.go
@@ -27,7 +27,9 @@ import (
"go.uber.org/multierr"
"go.uber.org/zap"
+ "k8s.io/apimachinery/pkg/util/wait"
+ "github.com/api7/ingress-controller/pkg/apisix/cache"
"github.com/api7/ingress-controller/pkg/log"
)
@@ -54,15 +56,17 @@ type ClusterOptions struct {
}
type cluster struct {
- name string
- baseURL string
- adminKey string
- cli *http.Client
-
- route Route
- upstream Upstream
- service Service
- ssl SSL
+ name string
+ baseURL string
+ adminKey string
+ cli *http.Client
+ cache cache.Cache
+ cacheSynced chan struct{}
+ cacheSyncErr error
+ route Route
+ upstream Upstream
+ service Service
+ ssl SSL
}
func newCluster(o *ClusterOptions) (Cluster, error) {
@@ -85,20 +89,141 @@ func newCluster(o *ClusterOptions) (Cluster, error) {
ExpectContinueTimeout: o.Timeout,
},
},
+ cache: nil,
+ cacheSynced: make(chan struct{}),
}
c.route = newRouteClient(c)
c.upstream = newUpstreamClient(c)
c.service = newServiceClient(c)
c.ssl = newSSLClient(c)
+ go c.syncCache()
+
return c, nil
}
-// String exposes the client information in human readable format.
+func (c *cluster) syncCache() {
+ log.Infow("syncing cache", zap.String("cluster", c.name))
+ now := time.Now()
+ defer func() {
+ if c.cacheSyncErr == nil {
+ log.Infow("cache synced",
+ zap.String("cost_time", time.Now().Sub(now).String()),
+ zap.String("cluster", c.name),
+ )
+ } else {
+ log.Errorw("failed to sync cache",
+ zap.String("cost_time", time.Now().Sub(now).String()),
+ zap.String("cluster", c.name),
+ )
+ }
+ }()
+
+ backoff := wait.Backoff{
+ Duration: time.Second,
+ Factor: 2,
+ Steps: 6,
+ }
+ err := wait.ExponentialBackoff(backoff, c.syncCacheOnce)
+ if err != nil {
+ c.cacheSyncErr = err
+ }
+ close(c.cacheSynced)
+}
+
+func (c *cluster) syncCacheOnce() (bool, error) {
+ dbcache, err := cache.NewMemDBCache()
+ if err != nil {
+ return false, err
+ }
+ c.cache = dbcache
+
+ routes, err := c.route.List(context.TODO())
+ if err != nil {
+ log.Errorf("failed to list route in APISIX: %s", err)
+ return false, err
+ }
+ services, err := c.service.List(context.TODO())
+ if err != nil {
+ log.Errorf("failed to list services in APISIX: %s", err)
+ return false, err
+ }
+ upstreams, err := c.upstream.List(context.TODO())
+ if err != nil {
+ log.Errorf("failed to list upstreams in APISIX: %s", err)
+ return false, err
+ }
+ ssl, err := c.ssl.List(context.TODO())
+ if err != nil {
+ log.Errorf("failed to list ssl in APISIX: %s", err)
+ return false, err
+ }
+
+ for _, r := range routes {
+ if err := c.cache.InsertRoute(r); err != nil {
+ log.Errorw("failed to insert route to cache",
+ zap.String("route", *r.ID),
+ zap.String("cluster", c.name),
+ zap.String("error", err.Error()),
+ )
+ return false, err
+ }
+ }
+ for _, s := range services {
+ if err := c.cache.InsertService(s); err != nil {
+ log.Errorw("failed to insert service to cache",
+ zap.String("service", *s.ID),
+ zap.String("cluster", c.name),
+ zap.String("error", err.Error()),
+ )
+ return false, err
+ }
+ }
+ for _, u := range upstreams {
+ if err := c.cache.InsertUpstream(u); err != nil {
+ log.Errorw("failed to insert upstream to cache",
+ zap.String("upstream", *u.ID),
+ zap.String("cluster", c.name),
+ zap.String("error", err.Error()),
+ )
+ return false, err
+ }
+ }
+ for _, s := range ssl {
+ if err := c.cache.InsertSSL(s); err != nil {
+ log.Errorw("failed to insert ssl to cache",
+ zap.String("ssl", *s.ID),
+ zap.String("cluster", c.name),
+ zap.String("error", err.Error()),
+ )
+ return false, err
+ }
+ }
+ return true, nil
+}
+
+// String implements Cluster.String method.
func (c *cluster) String() string {
return fmt.Sprintf("name=%s; base_url=%s", c.name, c.baseURL)
}
+// HasSynced implements Cluster.HasSynced method.
+func (c *cluster) HasSynced(ctx context.Context) error {
+ if c.cacheSyncErr != nil {
+ return c.cacheSyncErr
+ }
+ now := time.Now()
+ log.Warnf("waiting cluster %s to ready, it may takes a while", c.name)
+ select {
+ case <-ctx.Done():
+ log.Errorf("failed to wait cluster to ready: %s", ctx.Err())
+ return ctx.Err()
+ case <-c.cacheSynced:
+ log.Warnf("cluster %s now is ready, cost time %s", c.name, time.Now().Sub(now).String())
+ return nil
+ }
+}
+
// Route implements Cluster.Route method.
func (c *cluster) Route() Route {
return c.route
diff --git a/pkg/apisix/cluster_test.go b/pkg/apisix/cluster_test.go
index a803e31..0f7e569 100644
--- a/pkg/apisix/cluster_test.go
+++ b/pkg/apisix/cluster_test.go
@@ -24,7 +24,10 @@ import (
)
func TestAddCluster(t *testing.T) {
- apisix, err := NewForOptions(&ClusterOptions{
+ apisix, err := NewClient()
+ assert.Nil(t, err)
+
+ err = apisix.AddCluster(&ClusterOptions{
BaseURL: "http://service1:9080/apisix/admin",
})
assert.Nil(t, err)
@@ -49,7 +52,10 @@ func TestAddCluster(t *testing.T) {
}
func TestNonExistentCluster(t *testing.T) {
- apisix, err := NewForOptions(&ClusterOptions{
+ apisix, err := NewClient()
+ assert.Nil(t, err)
+
+ err = apisix.AddCluster(&ClusterOptions{
BaseURL: "http://service1:9080/apisix/admin",
})
assert.Nil(t, err)
diff --git a/pkg/apisix/nonexistentclient.go b/pkg/apisix/nonexistentclient.go
index 934eac4..ccd3abf 100644
--- a/pkg/apisix/nonexistentclient.go
+++ b/pkg/apisix/nonexistentclient.go
@@ -18,6 +18,7 @@ package apisix
import (
"context"
+ "github.com/api7/ingress-controller/pkg/apisix/cache"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
)
@@ -45,6 +46,10 @@ type embedDummyResourceImplementer struct {
type dummyRoute struct{}
+func (f *dummyRoute) Get(_ context.Context, _ string) (*v1.Route, error) {
+ return nil, ErrClusterNotExist
+}
+
func (f *dummyRoute) List(_ context.Context) ([]*v1.Route, error) {
return nil, ErrClusterNotExist
}
@@ -63,6 +68,10 @@ func (f *dummyRoute) Update(_ context.Context, _ *v1.Route) (*v1.Route, error) {
type dummySSL struct{}
+func (f *dummySSL) Get(_ context.Context, _ string) (*v1.Ssl, error) {
+ return nil, ErrClusterNotExist
+}
+
func (f *dummySSL) List(_ context.Context) ([]*v1.Ssl, error) {
return nil, ErrClusterNotExist
}
@@ -81,6 +90,10 @@ func (f *dummySSL) Update(_ context.Context, _ *v1.Ssl) (*v1.Ssl, error) {
type dummyUpstream struct{}
+func (f *dummyUpstream) Get(_ context.Context, _ string) (*v1.Upstream, error) {
+ return nil, ErrClusterNotExist
+}
+
func (f *dummyUpstream) List(_ context.Context) ([]*v1.Upstream, error) {
return nil, ErrClusterNotExist
}
@@ -99,6 +112,10 @@ func (f *dummyUpstream) Update(_ context.Context, _ *v1.Upstream) (*v1.Upstream,
type dummyService struct{}
+func (f *dummyService) Get(_ context.Context, _ string) (*v1.Service, error) {
+ return nil, ErrClusterNotExist
+}
+
func (f *dummyService) List(_ context.Context) ([]*v1.Service, error) {
return nil, ErrClusterNotExist
}
@@ -130,3 +147,32 @@ func (nc *nonExistentCluster) Service() Service {
func (nc *nonExistentCluster) Upstream() Upstream {
return nc.upstream
}
+
+func (nc *nonExistentCluster) HasSynced(_ context.Context) error {
+ return nil
+}
+
+func (nc *nonExistentCluster) String() string {
+ return "non-existent cluster"
+}
+
+type dummyCache struct{}
+
+var _ cache.Cache = &dummyCache{}
+
+func (c *dummyCache) InsertRoute(_ *v1.Route) error { return nil }
+func (c *dummyCache) InsertService(_ *v1.Service) error { return nil }
+func (c *dummyCache) InsertSSL(_ *v1.Ssl) error { return nil }
+func (c *dummyCache) InsertUpstream(_ *v1.Upstream) error { return nil }
+func (c *dummyCache) GetRoute(_ string) (*v1.Route, error) { return nil, cache.ErrNotFound }
+func (c *dummyCache) GetService(_ string) (*v1.Service, error) { return nil, cache.ErrNotFound }
+func (c *dummyCache) GetSSL(_ string) (*v1.Ssl, error) { return nil, cache.ErrNotFound }
+func (c *dummyCache) GetUpstream(_ string) (*v1.Upstream, error) { return nil, cache.ErrNotFound }
+func (c *dummyCache) ListRoutes() ([]*v1.Route, error) { return nil, nil }
+func (c *dummyCache) ListServices() ([]*v1.Service, error) { return nil, nil }
+func (c *dummyCache) ListSSL() ([]*v1.Ssl, error) { return nil, nil }
+func (c *dummyCache) ListUpstreams() ([]*v1.Upstream, error) { return nil, nil }
+func (c *dummyCache) DeleteRoute(_ *v1.Route) error { return nil }
+func (c *dummyCache) DeleteService(_ *v1.Service) error { return nil }
+func (c *dummyCache) DeleteSSL(_ *v1.Ssl) error { return nil }
+func (c *dummyCache) DeleteUpstream(_ *v1.Upstream) error { return nil }
diff --git a/pkg/apisix/route.go b/pkg/apisix/route.go
index a505d28..51d9c9e 100644
--- a/pkg/apisix/route.go
+++ b/pkg/apisix/route.go
@@ -22,6 +22,7 @@ import (
"go.uber.org/zap"
+ "github.com/api7/ingress-controller/pkg/apisix/cache"
"github.com/api7/ingress-controller/pkg/log"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
)
@@ -48,9 +49,48 @@ func newRouteClient(c *cluster) Route {
}
}
-func (r *routeClient) List(ctx context.Context) ([]*v1.Route, error) {
- log.Infow("try to list routes in APISIX", zap.String("url", r.url))
+func (r *routeClient) Get(ctx context.Context, fullname string) (*v1.Route, error) {
+ log.Infow("try to look up route",
+ zap.String("fullname", fullname),
+ zap.String("url", r.url),
+ zap.String("cluster", r.clusterName),
+ )
+ route, err := r.cluster.cache.GetRoute(fullname)
+ if err == nil {
+ return route, nil
+ }
+ if err != cache.ErrNotFound {
+ log.Errorw("failed to find route in cache, will try to lookup from APISIX",
+ zap.String("fullname", fullname),
+ zap.Error(err),
+ )
+ } else {
+ log.Warnw("failed to find route in cache, will try to lookup from APISIX",
+ zap.String("fullname", fullname),
+ zap.Error(err),
+ )
+ }
+
+ // FIXME Replace the List with accurate get since list is not trivial.
+ list, err := r.List(ctx)
+ if err != nil {
+ return nil, err
+ }
+ for _, elem := range list {
+ if *elem.FullName == fullname {
+ return elem, nil
+ }
+ }
+ return nil, cache.ErrNotFound
+}
+// List is only used in cache warming up. So here just pass through
+// to APISIX.
+func (r *routeClient) List(ctx context.Context) ([]*v1.Route, error) {
+ log.Infow("try to list routes in APISIX",
+ zap.String("cluster", r.clusterName),
+ zap.String("url", r.url),
+ )
routeItems, err := r.cluster.listResource(ctx, r.url)
if err != nil {
log.Errorf("failed to list routes: %s", err)
@@ -77,7 +117,16 @@ func (r *routeClient) List(ctx context.Context) ([]*v1.Route, error) {
}
func (r *routeClient) Create(ctx context.Context, obj *v1.Route) (*v1.Route, error) {
- log.Infow("try to create route", zap.String("host", *obj.Host))
+ log.Infow("try to create route",
+ zap.String("host", *obj.Host),
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", r.clusterName),
+ zap.String("url", r.url),
+ )
+
+ if err := r.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
data, err := json.Marshal(routeReqBody{
Desc: obj.Name,
URI: obj.Path,
@@ -101,17 +150,48 @@ func (r *routeClient) Create(ctx context.Context, obj *v1.Route) (*v1.Route, err
if obj.Group != nil {
clusterName = *obj.Group
}
- return resp.Item.route(clusterName)
+ route, err := resp.Item.route(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := r.cluster.cache.InsertRoute(route); err != nil {
+ log.Errorf("failed to reflect route create to cache: %s", err)
+ return nil, err
+ }
+ return route, nil
}
func (r *routeClient) Delete(ctx context.Context, obj *v1.Route) error {
- log.Infof("delete route, id:%s", *obj.ID)
+ log.Infow("try to delete route",
+ zap.String("id", *obj.ID),
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", r.clusterName),
+ zap.String("url", r.url),
+ )
+ if err := r.cluster.HasSynced(ctx); err != nil {
+ return err
+ }
url := r.url + "/" + *obj.ID
- return r.cluster.deleteResource(ctx, url)
+ if err := r.cluster.deleteResource(ctx, url); err != nil {
+ return err
+ }
+ if err := r.cluster.cache.DeleteRoute(obj); err != nil {
+ log.Errorf("failed to reflect route delete to cache: %s", err)
+ return err
+ }
+ return nil
}
func (r *routeClient) Update(ctx context.Context, obj *v1.Route) (*v1.Route, error) {
- log.Infof("update route, id:%s", *obj.ID)
+ log.Infow("try to update route",
+ zap.String("id", *obj.ID),
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", r.clusterName),
+ zap.String("url", r.url),
+ )
+ if err := r.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
body, err := json.Marshal(routeReqBody{
Desc: obj.Name,
Host: obj.Host,
@@ -132,5 +212,13 @@ func (r *routeClient) Update(ctx context.Context, obj *v1.Route) (*v1.Route, err
if obj.Group != nil {
clusterName = *obj.Group
}
- return resp.Item.route(clusterName)
+ route, err := resp.Item.route(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := r.cluster.cache.InsertRoute(route); err != nil {
+ log.Errorf("failed to reflect route update to cache: %s", err)
+ return nil, err
+ }
+ return route, nil
}
diff --git a/pkg/apisix/route_test.go b/pkg/apisix/route_test.go
index 7f1d6c1..c49b9c6 100644
--- a/pkg/apisix/route_test.go
+++ b/pkg/apisix/route_test.go
@@ -171,9 +171,13 @@ func TestRouteClient(t *testing.T) {
Path: "/apisix/admin",
}
+ closedCh := make(chan struct{})
+ close(closedCh)
cli := newRouteClient(&cluster{
- baseURL: u.String(),
- cli: http.DefaultClient,
+ baseURL: u.String(),
+ cli: http.DefaultClient,
+ cache: &dummyCache{},
+ cacheSynced: closedCh,
})
// Create
@@ -185,6 +189,7 @@ func TestRouteClient(t *testing.T) {
Host: &host,
Path: &uri,
Name: &name,
+ FullName: &name,
ServiceId: &id,
UpstreamId: &id,
})
@@ -195,6 +200,7 @@ func TestRouteClient(t *testing.T) {
Host: &host,
Path: &uri,
Name: &name,
+ FullName: &name,
ServiceId: &id,
UpstreamId: &id,
})
@@ -223,6 +229,7 @@ func TestRouteClient(t *testing.T) {
Host: &host,
Path: &uri,
Name: &name,
+ FullName: &name,
ServiceId: &id,
UpstreamId: &id,
})
diff --git a/pkg/apisix/service.go b/pkg/apisix/service.go
index 85c477f..a58f099 100644
--- a/pkg/apisix/service.go
+++ b/pkg/apisix/service.go
@@ -20,9 +20,11 @@ import (
"context"
"encoding/json"
+ "go.uber.org/zap"
+
+ "github.com/api7/ingress-controller/pkg/apisix/cache"
"github.com/api7/ingress-controller/pkg/log"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
- "go.uber.org/zap"
)
type serviceClient struct {
@@ -45,8 +47,48 @@ func newServiceClient(c *cluster) Service {
}
}
+func (s *serviceClient) Get(ctx context.Context, fullname string) (*v1.Service, error) {
+ log.Infow("try to look up service",
+ zap.String("fullname", fullname),
+ zap.String("url", s.url),
+ zap.String("cluster", s.clusterName),
+ )
+ svc, err := s.cluster.cache.GetService(fullname)
+ if err == nil {
+ return svc, nil
+ }
+ if err != cache.ErrNotFound {
+ log.Errorw("failed to find service in cache, will try to look up from APISIX",
+ zap.String("fullname", fullname),
+ zap.Error(err),
+ )
+ } else {
+ log.Warnw("failed to find service in cache, will try to look up from APISIX",
+ zap.String("fullname", fullname),
+ zap.Error(err),
+ )
+ }
+
+ // FIXME Replace the List with accurate get since list is not trivial.
+ list, err := s.List(ctx)
+ if err != nil {
+ return nil, err
+ }
+ for _, elem := range list {
+ if *elem.FullName == fullname {
+ return elem, nil
+ }
+ }
+ return nil, cache.ErrNotFound
+}
+
+// List is only used in cache warming up. So here just pass through
+// to APISIX.
func (s *serviceClient) List(ctx context.Context) ([]*v1.Service, error) {
- log.Infow("try to list services in APISIX", zap.String("url", s.url))
+ log.Infow("try to list services in APISIX",
+ zap.String("url", s.url),
+ zap.String("cluster", s.clusterName),
+ )
upsItems, err := s.cluster.listResource(ctx, s.url)
if err != nil {
@@ -72,7 +114,14 @@ func (s *serviceClient) List(ctx context.Context) ([]*v1.Service, error) {
}
func (s *serviceClient) Create(ctx context.Context, obj *v1.Service) (*v1.Service, error) {
- log.Infow("try to create service", zap.String("full_name", *obj.FullName))
+ log.Infow("try to create service",
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", s.clusterName),
+ zap.String("url", s.url),
+ )
+ if err := s.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
body, err := json.Marshal(serviceItem{
UpstreamId: obj.UpstreamId,
@@ -93,17 +142,49 @@ func (s *serviceClient) Create(ctx context.Context, obj *v1.Service) (*v1.Servic
if obj.Group != nil {
clusterName = *obj.Group
}
- return resp.Item.service(clusterName)
+ svc, err := resp.Item.service(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := s.cluster.cache.InsertService(svc); err != nil {
+ log.Errorf("failed to reflect service create to cache: %s", err)
+ return nil, err
+ }
+ return svc, nil
}
func (s *serviceClient) Delete(ctx context.Context, obj *v1.Service) error {
- log.Infof("delete service, id:%s", *obj.ID)
+ log.Infow("try to delete service",
+ zap.String("id", *obj.ID),
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", s.clusterName),
+ zap.String("url", s.url),
+ )
+ if err := s.cluster.HasSynced(ctx); err != nil {
+ return err
+ }
url := s.url + "/" + *obj.ID
- return s.cluster.deleteResource(ctx, url)
+ if err := s.cluster.deleteResource(ctx, url); err != nil {
+ return err
+ }
+ if err := s.cluster.cache.DeleteService(obj); err != nil {
+ log.Errorf("failed to reflect service delete to cache: %s", err)
+ return err
+ }
+ return nil
}
func (s *serviceClient) Update(ctx context.Context, obj *v1.Service) (*v1.Service, error) {
- log.Infof("update service, id:%s", *obj.ID)
+ log.Infow("try to update service",
+ zap.String("id", *obj.ID),
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", s.clusterName),
+ zap.String("url", s.url),
+ )
+
+ if err := s.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
body, err := json.Marshal(serviceItem{
UpstreamId: obj.UpstreamId,
@@ -124,5 +205,13 @@ func (s *serviceClient) Update(ctx context.Context, obj *v1.Service) (*v1.Servic
if obj.Group != nil {
clusterName = *obj.Group
}
- return resp.Item.service(clusterName)
+ svc, err := resp.Item.service(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := s.cluster.cache.InsertService(obj); err != nil {
+ log.Errorf("failed to reflect service update to cache: %s", err)
+ return nil, err
+ }
+ return svc, nil
}
diff --git a/pkg/apisix/service_test.go b/pkg/apisix/service_test.go
index a2ed175..a7054e5 100644
--- a/pkg/apisix/service_test.go
+++ b/pkg/apisix/service_test.go
@@ -148,9 +148,13 @@ func TestServiceClient(t *testing.T) {
Host: srv.Addr,
Path: "/apisix/admin",
}
+ closedCh := make(chan struct{})
+ close(closedCh)
cli := newServiceClient(&cluster{
- baseURL: u.String(),
- cli: http.DefaultClient,
+ baseURL: u.String(),
+ cli: http.DefaultClient,
+ cache: &dummyCache{},
+ cacheSynced: closedCh,
})
// Create
diff --git a/pkg/apisix/ssl.go b/pkg/apisix/ssl.go
index d904e6d..8d0ee7e 100644
--- a/pkg/apisix/ssl.go
+++ b/pkg/apisix/ssl.go
@@ -20,9 +20,9 @@ import (
"context"
"encoding/json"
- "github.com/api7/ingress-controller/pkg/log"
"go.uber.org/zap"
+ "github.com/api7/ingress-controller/pkg/log"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
)
@@ -40,8 +40,23 @@ func newSSLClient(c *cluster) SSL {
}
}
+func (s *sslClient) Get(_ context.Context, fullname string) (*v1.Ssl, error) {
+ log.Infow("try to look up ssl",
+ zap.String("fullname", fullname),
+ zap.String("url", s.url),
+ zap.String("cluster", s.clusterName),
+ )
+
+ return s.cluster.cache.GetSSL(fullname)
+}
+
+// List is only used in cache warming up. So here just pass through
+// to APISIX.
func (s *sslClient) List(ctx context.Context) ([]*v1.Ssl, error) {
- log.Infow("try to list ssl in APISIX", zap.String("url", s.url))
+ log.Infow("try to list ssl in APISIX",
+ zap.String("url", s.url),
+ zap.String("cluster", s.clusterName),
+ )
sslItems, err := s.cluster.listResource(ctx, s.url)
if err != nil {
@@ -68,7 +83,13 @@ func (s *sslClient) List(ctx context.Context) ([]*v1.Ssl, error) {
}
func (s *sslClient) Create(ctx context.Context, obj *v1.Ssl) (*v1.Ssl, error) {
- log.Info("try to create ssl")
+ log.Infow("try to create ssl",
+ zap.String("cluster", s.clusterName),
+ zap.String("url", s.url),
+ )
+ if err := s.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
data, err := json.Marshal(v1.Ssl{
Snis: obj.Snis,
Cert: obj.Cert,
@@ -90,17 +111,46 @@ func (s *sslClient) Create(ctx context.Context, obj *v1.Ssl) (*v1.Ssl, error) {
clusterName = *obj.Group
}
- return resp.Item.ssl(clusterName)
+ ssl, err := resp.Item.ssl(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := s.cluster.cache.InsertSSL(ssl); err != nil {
+ log.Errorf("failed to reflect ssl create to cache: %s", err)
+ return nil, err
+ }
+ return ssl, nil
}
func (s *sslClient) Delete(ctx context.Context, obj *v1.Ssl) error {
- log.Infof("delete ssl, id:%s", *obj.ID)
+ log.Infow("try to delete ssl",
+ zap.String("id", *obj.ID),
+ zap.String("cluster", s.clusterName),
+ zap.String("url", s.url),
+ )
+ if err := s.cluster.HasSynced(ctx); err != nil {
+ return err
+ }
url := s.url + "/" + *obj.ID
- return s.cluster.deleteResource(ctx, url)
+ if err := s.cluster.deleteResource(ctx, url); err != nil {
+ return err
+ }
+ if err := s.cluster.cache.DeleteSSL(obj); err != nil {
+ log.Errorf("failed to reflect ssl delete to cache: %s", err)
+ return err
+ }
+ return nil
}
func (s *sslClient) Update(ctx context.Context, obj *v1.Ssl) (*v1.Ssl, error) {
- log.Infof("update ssl, id:%s", *obj.ID)
+ log.Infow("try to update ssl",
+ zap.String("id", *obj.ID),
+ zap.String("cluster", s.clusterName),
+ zap.String("url", s.url),
+ )
+ if err := s.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
url := s.url + "/" + *obj.ID
data, err := json.Marshal(v1.Ssl{
ID: obj.ID,
@@ -121,5 +171,14 @@ func (s *sslClient) Update(ctx context.Context, obj *v1.Ssl) (*v1.Ssl, error) {
if obj.Group != nil {
clusterName = *obj.Group
}
- return resp.Item.ssl(clusterName)
+ ssl, err := resp.Item.ssl(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := s.cluster.cache.InsertSSL(ssl); err != nil {
+ log.Errorf("failed to reflect ssl update to cache: %s", err)
+ return nil, err
+
+ }
+ return ssl, nil
}
diff --git a/pkg/apisix/ssl_test.go b/pkg/apisix/ssl_test.go
index e176b1e..8ed9b1a 100644
--- a/pkg/apisix/ssl_test.go
+++ b/pkg/apisix/ssl_test.go
@@ -148,9 +148,13 @@ func TestSSLClient(t *testing.T) {
Host: srv.Addr,
Path: "/apisix/admin",
}
+ closedCh := make(chan struct{})
+ close(closedCh)
cli := newSSLClient(&cluster{
- baseURL: u.String(),
- cli: http.DefaultClient,
+ baseURL: u.String(),
+ cli: http.DefaultClient,
+ cache: &dummyCache{},
+ cacheSynced: closedCh,
})
// Create
diff --git a/pkg/apisix/upstream.go b/pkg/apisix/upstream.go
index 0fbccf3..11d0b94 100644
--- a/pkg/apisix/upstream.go
+++ b/pkg/apisix/upstream.go
@@ -22,6 +22,7 @@ import (
"go.uber.org/zap"
+ "github.com/api7/ingress-controller/pkg/apisix/cache"
"github.com/api7/ingress-controller/pkg/log"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
)
@@ -81,8 +82,48 @@ func newUpstreamClient(c *cluster) Upstream {
}
}
+func (u *upstreamClient) Get(ctx context.Context, fullname string) (*v1.Upstream, error) {
+ log.Infow("try to look up upstream",
+ zap.String("fullname", fullname),
+ zap.String("url", u.url),
+ zap.String("cluster", u.clusterName),
+ )
+ ups, err := u.cluster.cache.GetUpstream(fullname)
+ if err == nil {
+ return ups, nil
+ }
+ if err != cache.ErrNotFound {
+ log.Errorw("failed to find upstream in cache, will try to lookup from APISIX",
+ zap.String("fullname", fullname),
+ zap.Error(err),
+ )
+ } else {
+ log.Warnw("failed to find upstream in cache, will try to lookup from APISIX",
+ zap.String("fullname", fullname),
+ zap.Error(err),
+ )
+ }
+
+ // FIXME Replace the List with accurate get since list is not trivial.
+ list, err := u.List(ctx)
+ if err != nil {
+ return nil, err
+ }
+ for _, elem := range list {
+ if *elem.FullName == fullname {
+ return elem, nil
+ }
+ }
+ return nil, cache.ErrNotFound
+}
+
+// List is only used in cache warming up. So here just pass through
+// to APISIX.
func (u *upstreamClient) List(ctx context.Context) ([]*v1.Upstream, error) {
- log.Infow("try to list upstreams in APISIX", zap.String("url", u.url))
+ log.Infow("try to list upstreams in APISIX",
+ zap.String("url", u.url),
+ zap.String("cluster", u.clusterName),
+ )
upsItems, err := u.cluster.listResource(ctx, u.url)
if err != nil {
@@ -102,16 +143,22 @@ func (u *upstreamClient) List(ctx context.Context) ([]*v1.Upstream, error) {
return nil, err
}
items = append(items, ups)
- log.Infof("list upstream #%d, body: %s", i, string(item.Value))
+ log.Debugf("list upstream #%d, body: %s", i, string(item.Value))
}
return items, nil
}
func (u *upstreamClient) Create(ctx context.Context, obj *v1.Upstream) (*v1.Upstream, error) {
log.Infow("try to create upstream",
- zap.String("full_name", *obj.FullName),
+ zap.String("fullname", *obj.FullName),
+ zap.String("url", u.url),
+ zap.String("cluster", u.clusterName),
)
+ if err := u.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
+
nodes := make(upstreamNodes, 0, len(obj.Nodes))
for _, node := range obj.Nodes {
nodes = append(nodes, upstreamNode{
@@ -141,17 +188,50 @@ func (u *upstreamClient) Create(ctx context.Context, obj *v1.Upstream) (*v1.Upst
if obj.Group != nil {
clusterName = *obj.Group
}
- return resp.Item.upstream(clusterName)
+ ups, err := resp.Item.upstream(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := u.cluster.cache.InsertUpstream(ups); err != nil {
+ log.Errorf("failed to reflect upstream create to cache: %s", err)
+ return nil, err
+ }
+ return ups, err
}
func (u *upstreamClient) Delete(ctx context.Context, obj *v1.Upstream) error {
- log.Infof("delete upstream, id:%s", *obj.ID)
+ log.Infow("try to delete upstream",
+ zap.String("id", *obj.ID),
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", u.clusterName),
+ zap.String("url", u.url),
+ )
+
+ if err := u.cluster.HasSynced(ctx); err != nil {
+ return err
+ }
url := u.url + "/" + *obj.ID
- return u.cluster.deleteResource(ctx, url)
+ if err := u.cluster.deleteResource(ctx, url); err != nil {
+ return err
+ }
+ if err := u.cluster.cache.DeleteUpstream(obj); err != nil {
+ log.Errorf("failed to reflect upstream delete to cache: %s", err.Error())
+ return err
+ }
+ return nil
}
func (u *upstreamClient) Update(ctx context.Context, obj *v1.Upstream) (*v1.Upstream, error) {
- log.Infof("update upstream, id:%s", *obj.ID)
+ log.Infow("try to update upstream",
+ zap.String("id", *obj.ID),
+ zap.String("fullname", *obj.FullName),
+ zap.String("cluster", u.clusterName),
+ zap.String("url", u.url),
+ )
+
+ if err := u.cluster.HasSynced(ctx); err != nil {
+ return nil, err
+ }
nodes := make(upstreamNodes, 0, len(obj.Nodes))
for _, node := range obj.Nodes {
@@ -173,7 +253,7 @@ func (u *upstreamClient) Update(ctx context.Context, obj *v1.Upstream) (*v1.Upst
}
url := u.url + "/" + *obj.ID
- log.Infow("upating upstream", zap.ByteString("body", body), zap.String("url", url))
+ log.Infow("updating upstream", zap.ByteString("body", body), zap.String("url", url))
resp, err := u.cluster.updateResource(ctx, url, bytes.NewReader(body))
if err != nil {
return nil, err
@@ -182,5 +262,13 @@ func (u *upstreamClient) Update(ctx context.Context, obj *v1.Upstream) (*v1.Upst
if obj.Group != nil {
clusterName = *obj.Group
}
- return resp.Item.upstream(clusterName)
+ ups, err := resp.Item.upstream(clusterName)
+ if err != nil {
+ return nil, err
+ }
+ if err := u.cluster.cache.InsertUpstream(ups); err != nil {
+ log.Errorf("failed to reflect upstraem update to cache: %s", err)
+ return nil, err
+ }
+ return ups, err
}
diff --git a/pkg/apisix/upstream_test.go b/pkg/apisix/upstream_test.go
index 63bed5c..fef4901 100644
--- a/pkg/apisix/upstream_test.go
+++ b/pkg/apisix/upstream_test.go
@@ -149,9 +149,13 @@ func TestUpstreamClient(t *testing.T) {
Host: srv.Addr,
Path: "/apisix/admin",
}
+ closedCh := make(chan struct{})
+ close(closedCh)
cli := newUpstreamClient(&cluster{
- baseURL: u.String(),
- cli: http.DefaultClient,
+ baseURL: u.String(),
+ cli: http.DefaultClient,
+ cache: &dummyCache{},
+ cacheSynced: closedCh,
})
// Create
diff --git a/pkg/ingress/apisix/tls_test.go b/pkg/ingress/apisix/tls_test.go
index b61223a..ad58c41 100644
--- a/pkg/ingress/apisix/tls_test.go
+++ b/pkg/ingress/apisix/tls_test.go
@@ -163,7 +163,9 @@ func (sc *SecretClientErrorMock) FindByName(namespace, name string) (*v1.Secret,
}
func setDummyApisixClient(t *testing.T) {
- cli, err := apisixhttp.NewForOptions(&apisixhttp.ClusterOptions{
+ cli, err := apisixhttp.NewClient()
+ assert.Nil(t, err)
+ err = cli.AddCluster(&apisixhttp.ClusterOptions{
Name: "",
BaseURL: "http://127.0.0.2:9080/apisix/admin",
})
diff --git a/pkg/ingress/controller/apisix_route.go b/pkg/ingress/controller/apisix_route.go
index c9938e7..73fc010 100644
--- a/pkg/ingress/controller/apisix_route.go
+++ b/pkg/ingress/controller/apisix_route.go
@@ -213,7 +213,8 @@ func (c *ApisixRouteController) add(key string) error {
log.Infof("apisixRoute %s is removed", key)
return nil
}
- runtime.HandleError(fmt.Errorf("failed to list apisixRoute %s/%s", key, err.Error()))
+ log.Errorf("failed to list ApisixRoute %s: %s", key, err.Error())
+ runtime.HandleError(fmt.Errorf("failed to list ApisixRoute %s: %s", key, err.Error()))
return err
}
apisixRoute := apisix.ApisixRoute(*apisixIngressRoute)
diff --git a/pkg/ingress/controller/controller.go b/pkg/ingress/controller/controller.go
index 14fef47..7949dc4 100644
--- a/pkg/ingress/controller/controller.go
+++ b/pkg/ingress/controller/controller.go
@@ -56,6 +56,7 @@ type Controller struct {
cfg *config.Config
wg sync.WaitGroup
watchingNamespace map[string]struct{}
+ apisix apisix.APISIX
apiServer *api.Server
clientset kubernetes.Interface
crdClientset crdclientset.Interface
@@ -71,12 +72,7 @@ func NewController(cfg *config.Config) (*Controller, error) {
if podNamespace == "" {
podNamespace = "default"
}
-
- client, err := apisix.NewForOptions(&apisix.ClusterOptions{
- Name: "",
- AdminKey: cfg.APISIX.AdminKey,
- BaseURL: cfg.APISIX.BaseURL,
- })
+ client, err := apisix.NewClient()
if err != nil {
return nil, err
}
@@ -107,6 +103,7 @@ func NewController(cfg *config.Config) (*Controller, error) {
namespace: podNamespace,
cfg: cfg,
apiServer: apiSrv,
+ apisix: client,
metricsCollector: metrics.NewPrometheusCollector(podName, podNamespace),
clientset: kube.GetKubeClient(),
crdClientset: crdClientset,
@@ -208,6 +205,22 @@ func (c *Controller) run(ctx context.Context) {
)
c.metricsCollector.ResetLeader(true)
+ err := c.apisix.AddCluster(&apisix.ClusterOptions{
+ Name: "",
+ AdminKey: c.cfg.APISIX.AdminKey,
+ BaseURL: c.cfg.APISIX.BaseURL,
+ })
+ if err != nil {
+ // TODO give up the leader role.
+ log.Errorf("failed to add default cluster: %s", err)
+ return
+ }
+ if err := c.apisix.Cluster("").HasSynced(ctx); err != nil {
+ // TODO give up the leader role.
+ log.Errorf("failed to wait the default cluster to be ready: %s", err)
+ return
+ }
+
ac := &Api6Controller{
KubeClientSet: c.clientset,
Api6ClientSet: c.crdClientset,
diff --git a/pkg/seven/apisix/client.go b/pkg/seven/apisix/client.go
deleted file mode 100644
index fcb2466..0000000
--- a/pkg/seven/apisix/client.go
+++ /dev/null
@@ -1,42 +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 (
- "fmt"
- "net/http"
- "time"
-
- "gopkg.in/resty.v1"
-)
-
-const (
- timeout = 3000
-)
-
-func Get(url string) ([]byte, error) {
- r := resty.New().
- SetTimeout(time.Duration(timeout)*time.Millisecond).
- R().
- SetHeader("content-type", "application/json")
- resp, err := r.Get(url)
- if err != nil {
- return nil, err
- }
- if resp.StatusCode() != http.StatusOK {
- return nil, fmt.Errorf("status: %d, body: %s", resp.StatusCode(), resp.Body())
- }
- return resp.Body(), nil
-}
diff --git a/pkg/seven/apisix/error.go b/pkg/seven/apisix/error.go
deleted file mode 100644
index d832854..0000000
--- a/pkg/seven/apisix/error.go
+++ /dev/null
@@ -1,22 +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
-
-type ErrorResponse struct {
- Cause string `json:"cause"`
- Index int64 `json:"index"`
- ErrorCode int64 `json:"errorCode"`
- Message string `json:"message"`
-}
diff --git a/pkg/seven/apisix/route.go b/pkg/seven/apisix/route.go
deleted file mode 100644
index 0e0712c..0000000
--- a/pkg/seven/apisix/route.go
+++ /dev/null
@@ -1,55 +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 (
- "context"
- "fmt"
-
- "github.com/api7/ingress-controller/pkg/seven/conf"
- sevendb "github.com/api7/ingress-controller/pkg/seven/db"
- "github.com/api7/ingress-controller/pkg/seven/utils"
- v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
-)
-
-// FindCurrentRoute find current route in memDB
-func FindCurrentRoute(route *v1.Route) (*v1.Route, error) {
- var cluster string
- if route.Group != nil {
- cluster = *route.Group
- }
- db := &sevendb.RouteRequest{Group: *route.Group, Name: *route.Name, FullName: *route.FullName}
- currentRoute, _ := db.FindByName()
- if currentRoute != nil {
- return currentRoute, nil
- } else {
- // find from apisix
- if routes, err := conf.Client.Cluster(cluster).Route().List(context.TODO()); err != nil {
- return nil, fmt.Errorf("list routes from etcd failed, err: %+v", err)
- } else {
- for _, r := range routes {
- if r.Name != nil && *r.Name == *route.Name {
- // insert to memDB
- db := &sevendb.RouteDB{Routes: []*v1.Route{r}}
- db.Insert()
- // return
- return r, nil
- }
- }
- }
-
- }
- return nil, utils.ErrNotFound
-}
diff --git a/pkg/seven/apisix/service.go b/pkg/seven/apisix/service.go
deleted file mode 100644
index 90f0388..0000000
--- a/pkg/seven/apisix/service.go
+++ /dev/null
@@ -1,53 +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 (
- "context"
- "fmt"
-
- "github.com/golang/glog"
-
- "github.com/api7/ingress-controller/pkg/seven/conf"
- sevendb "github.com/api7/ingress-controller/pkg/seven/db"
- v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
-)
-
-// FindCurrentService find service from memDB,
-// if Not Found, find service from apisix
-func FindCurrentService(group, name, fullName string) (*v1.Service, error) {
- db := sevendb.ServiceRequest{Group: group, Name: name, FullName: fullName}
- currentService, _ := db.FindByName()
- if currentService != nil {
- return currentService, nil
- } else {
- // find service from apisix
- if services, err := conf.Client.Cluster(group).Service().List(context.TODO()); err != nil {
- glog.Errorf("list services in etcd failed, group: %s, err: %+v", group, err)
- return nil, fmt.Errorf("list services failed, err: %+v", err)
- } else {
- for _, s := range services {
- if s.Name != nil && *(s.Name) == name {
- // and save to memDB
- db := &sevendb.ServiceDB{Services: []*v1.Service{s}}
- db.Insert()
- // return
- return s, nil
- }
- }
- }
- }
- return nil, nil
-}
diff --git a/pkg/seven/apisix/upstream.go b/pkg/seven/apisix/upstream.go
deleted file mode 100644
index 65c96cf..0000000
--- a/pkg/seven/apisix/upstream.go
+++ /dev/null
@@ -1,70 +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 (
- "context"
- "fmt"
-
- "github.com/golang/glog"
-
- "github.com/api7/ingress-controller/pkg/seven/conf"
- "github.com/api7/ingress-controller/pkg/seven/db"
- v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
-)
-
-// FindCurrentUpstream find upstream from memDB,
-// if Not Found, find upstream from apisix
-func FindCurrentUpstream(group, name, fullName string) (*v1.Upstream, error) {
- ur := &db.UpstreamRequest{Group: group, Name: name, FullName: fullName}
- currentUpstream, _ := ur.FindByName()
- if currentUpstream != nil {
- return currentUpstream, nil
- } else {
- // find upstream from apisix
- if upstreams, err := conf.Client.Cluster(group).Upstream().List(context.TODO()); err != nil {
- glog.Errorf("list upstreams in etcd failed, group: %s, err: %+v", group, err)
- return nil, fmt.Errorf("list upstreams failed, err: %+v", err)
- } else {
- for _, upstream := range upstreams {
- if upstream.Name != nil && *(upstream.Name) == name {
- // and save to memDB
- upstreamDB := &db.UpstreamDB{Upstreams: []*v1.Upstream{upstream}}
- upstreamDB.InsertUpstreams()
- //InsertUpstreams([]*v1.Upstream{upstream})
- // return
- return upstream, nil
- }
- }
- }
-
- }
- return nil, nil
-}
-
-func PatchNodes(upstream *v1.Upstream, nodes []*v1.Node) error {
- oldNodes := upstream.Nodes
- upstream.Nodes = nodes
- defer func() {
- // Restore it
- upstream.Nodes = oldNodes
- }()
- var cluster string
- if upstream.Group != nil {
- cluster = *upstream.Group
- }
- _, err := conf.Client.Cluster(cluster).Upstream().Update(context.TODO(), upstream)
- return err
-}
diff --git a/pkg/seven/db/route.go b/pkg/seven/db/route.go
deleted file mode 100644
index b8fbd4c..0000000
--- a/pkg/seven/db/route.go
+++ /dev/null
@@ -1,130 +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 db
-
-import (
- "github.com/hashicorp/go-memdb"
-
- "github.com/api7/ingress-controller/pkg/seven/utils"
- "github.com/api7/ingress-controller/pkg/types/apisix/v1"
-)
-
-const (
- Route = "Route"
-)
-
-type RouteRequest struct {
- Group string
- Name string
- FullName string
- ServiceId string
-}
-
-func (rr *RouteRequest) FindByName() (*v1.Route, error) {
- txn := DB.Txn(false)
- defer txn.Abort()
- if raw, err := txn.First(Route, "id", rr.FullName); err != nil {
- return nil, err
- } else {
- if raw != nil {
- currentRoute := raw.(*v1.Route)
- return currentRoute, nil
- }
- return nil, utils.ErrNotFound
- }
-}
-
-type RouteDB struct {
- Routes []*v1.Route
-}
-
-// InsertRoute insert route to memDB
-func (db *RouteDB) Insert() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, r := range db.Routes {
- if err := txn.Insert(Route, r); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-func (db *RouteDB) UpdateRoute() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, r := range db.Routes {
- // 1. delete
- if _, err := txn.DeleteAll(Route, "id", *(r.FullName)); err != nil {
- return err
- }
- // 2. insert
- if err := txn.Insert(Route, r); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-func (db *RouteDB) DeleteRoute() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, r := range db.Routes {
- //if _, err := txn.DeleteAll(Route, "id", *(r.ID)); err != nil {
- if _, err := txn.DeleteAll(Route, "id", *(r.FullName)); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-func (rr *RouteRequest) ExistByServiceId() (*v1.Route, error) {
- txn := DB.Txn(false)
- defer txn.Abort()
- if raw, err := txn.First(Route, "service_id", rr.ServiceId); err != nil {
- return nil, err
- } else {
- if raw != nil {
- firstRoute := raw.(*v1.Route)
- return firstRoute, nil
- }
- return nil, utils.ErrNotFound
- }
-}
-
-var routeSchema = &memdb.TableSchema{
- Name: Route,
- Indexes: map[string]*memdb.IndexSchema{
- "id": {
- Name: "id",
- Unique: true,
- Indexer: &memdb.StringFieldIndex{Field: "FullName"},
- },
- "name": {
- Name: "name",
- Unique: true,
- Indexer: &memdb.StringFieldIndex{Field: "Name"},
- AllowMissing: true,
- },
- "service_id": {
- Name: "service_id",
- Unique: false,
- Indexer: &memdb.StringFieldIndex{Field: "ServiceId"},
- },
- },
-}
diff --git a/pkg/seven/db/service.go b/pkg/seven/db/service.go
deleted file mode 100644
index 2928959..0000000
--- a/pkg/seven/db/service.go
+++ /dev/null
@@ -1,130 +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 db
-
-import (
- "github.com/hashicorp/go-memdb"
-
- "github.com/api7/ingress-controller/pkg/seven/utils"
- "github.com/api7/ingress-controller/pkg/types/apisix/v1"
-)
-
-const (
- Service = "Service"
-)
-
-type ServiceRequest struct {
- Group string
- Name string
- FullName string
- UpstreamId string
-}
-
-func (sr *ServiceRequest) FindByName() (*v1.Service, error) {
- txn := DB.Txn(false)
- defer txn.Abort()
- if raw, err := txn.First(Service, "id", sr.FullName); err != nil {
- return nil, err
- } else {
- if raw != nil {
- currentService := raw.(*v1.Service)
- return currentService, nil
- }
- return nil, utils.ErrNotFound
- }
-}
-
-func (db *ServiceDB) Insert() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, s := range db.Services {
- if err := txn.Insert(Service, s); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-type ServiceDB struct {
- Services []*v1.Service
-}
-
-func (db *ServiceDB) UpdateService() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, s := range db.Services {
- // 1. delete
- if _, err := txn.DeleteAll(Service, "id", *(s.FullName)); err != nil {
- return err
- }
- // 2. insert
- if err := txn.Insert(Service, s); err != nil {
- return err
- }
- }
-
- txn.Commit()
- return nil
-}
-
-func (db *ServiceDB) DeleteService() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, r := range db.Services {
- if _, err := txn.DeleteAll(Service, "id", *(r.FullName)); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-func (rr *ServiceRequest) ExistByUpstreamId() (*v1.Service, error) {
- txn := DB.Txn(false)
- defer txn.Abort()
- if raw, err := txn.First(Service, "upstream_id", rr.UpstreamId); err != nil {
- return nil, err
- } else {
- if raw != nil {
- firstService := raw.(*v1.Service)
- return firstService, nil
- }
- return nil, utils.ErrNotFound
- }
-}
-
-var serviceSchema = &memdb.TableSchema{
- Name: Service,
- Indexes: map[string]*memdb.IndexSchema{
- "id": {
- Name: "id",
- Unique: true,
- Indexer: &memdb.StringFieldIndex{Field: "FullName"},
- },
- "name": {
- Name: "name",
- Unique: true,
- Indexer: &memdb.StringFieldIndex{Field: "Name"},
- AllowMissing: true,
- },
- "upstream_id": {
- Name: "upstream_id",
- Unique: false,
- Indexer: &memdb.StringFieldIndex{Field: "UpstreamId"},
- AllowMissing: true,
- },
- },
-}
diff --git a/pkg/seven/db/store.go b/pkg/seven/db/store.go
deleted file mode 100644
index 9d71355..0000000
--- a/pkg/seven/db/store.go
+++ /dev/null
@@ -1,43 +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 db
-
-import "github.com/hashicorp/go-memdb"
-
-var DB *memdb.MemDB
-
-func init() {
- if db, err := NewDB(); err != nil {
- panic(err)
- } else {
- DB = db
- }
-}
-
-func NewDB() (*memdb.MemDB, error) {
- var schema = &memdb.DBSchema{
- Tables: map[string]*memdb.TableSchema{
- Service: serviceSchema,
- Route: routeSchema,
- Upstream: upstreamSchema,
- },
- }
-
- if memDB, err := memdb.NewMemDB(schema); err != nil {
- return nil, err
- } else {
- return memDB, nil
- }
-}
diff --git a/pkg/seven/db/upstream.go b/pkg/seven/db/upstream.go
deleted file mode 100644
index ba1b151..0000000
--- a/pkg/seven/db/upstream.go
+++ /dev/null
@@ -1,116 +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 db
-
-import (
- "github.com/hashicorp/go-memdb"
-
- "github.com/api7/ingress-controller/pkg/seven/utils"
- "github.com/api7/ingress-controller/pkg/types/apisix/v1"
-)
-
-const (
- Upstream = "Upstream"
-)
-
-type UpstreamDB struct {
- Upstreams []*v1.Upstream
-}
-
-type UpstreamRequest struct {
- Group string
- Name string
- FullName string
-}
-
-func (ur *UpstreamRequest) FindByName() (*v1.Upstream, error) {
- txn := DB.Txn(false)
- defer txn.Abort()
- if raw, err := txn.First(Upstream, "id", ur.FullName); err != nil {
- return nil, err
- } else {
- if raw != nil {
- currentUpstream := raw.(*v1.Upstream)
- return currentUpstream, nil
- }
- return nil, utils.ErrNotFound
- }
-}
-
-// insertUpstream insert upstream to memDB
-func (upstreamDB *UpstreamDB) InsertUpstreams() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, u := range upstreamDB.Upstreams {
- if err := txn.Insert(Upstream, u); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-func (upstreamDB *UpstreamDB) UpdateUpstreams() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, u := range upstreamDB.Upstreams {
- // delete
- if _, err := txn.DeleteAll(Upstream, "id", *(u.FullName)); err != nil {
- return err
- }
- // insert
- if err := txn.Insert(Upstream, u); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-func (db *UpstreamDB) DeleteUpstream() error {
- txn := DB.Txn(true)
- defer txn.Abort()
- for _, r := range db.Upstreams {
- if _, err := txn.DeleteAll(Upstream, "id", *(r.FullName)); err != nil {
- return err
- }
- }
- txn.Commit()
- return nil
-}
-
-var upstreamSchema = &memdb.TableSchema{
- Name: Upstream,
- Indexes: map[string]*memdb.IndexSchema{
- "id": {
- Name: "id",
- Unique: true,
- Indexer: &memdb.StringFieldIndex{Field: "FullName"},
- },
- "name": {
- Name: "name",
- Unique: true,
- Indexer: &memdb.StringFieldIndex{Field: "Name"},
- AllowMissing: true,
- },
- },
-}
-
-//func indexer() *memdb.CompoundMultiIndex{
-// var idx = make([]memdb.Indexer, 0)
-// idx = append(idx, &memdb.StringFieldIndex{Field: "Group"})
-// idx = append(idx, &memdb.StringFieldIndex{Field: "Name"})
-// return &memdb.CompoundMultiIndex{Indexes: idx, AllowMissing: false}
-//}
diff --git a/pkg/seven/state/builder.go b/pkg/seven/state/builder.go
index 0a90291..73fce0d 100644
--- a/pkg/seven/state/builder.go
+++ b/pkg/seven/state/builder.go
@@ -20,12 +20,9 @@ import (
"strconv"
"sync"
- "github.com/golang/glog"
-
+ "github.com/api7/ingress-controller/pkg/apisix/cache"
"github.com/api7/ingress-controller/pkg/log"
- "github.com/api7/ingress-controller/pkg/seven/apisix"
"github.com/api7/ingress-controller/pkg/seven/conf"
- "github.com/api7/ingress-controller/pkg/seven/db"
"github.com/api7/ingress-controller/pkg/seven/utils"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
)
@@ -115,8 +112,12 @@ func (r *routeWorker) trigger(event Event) {
log.Infof("trigger routeWorker %s from %s, %s", *r.Name, event.Op, *service.Name)
// padding
- currentRoute, err := apisix.FindCurrentRoute(r.Route)
- if err != nil && !errors.Is(err, utils.ErrNotFound) {
+ var cluster string
+ if r.Route.Group != nil {
+ cluster = *r.Route.Group
+ }
+ currentRoute, err := conf.Client.Cluster(cluster).Route().Get(context.TODO(), *r.Route.FullName)
+ if err != nil && !errors.Is(err, cache.ErrNotFound) {
errNotify = err
return
}
@@ -144,20 +145,12 @@ func (r *routeWorker) sync() error {
cluster = *r.Group
}
if *r.Route.ID != strconv.Itoa(0) {
- // 1. sync memDB
- db := &db.RouteDB{Routes: []*v1.Route{r.Route}}
- if err := db.UpdateRoute(); err != nil {
- log.Errorf("update route failed, route: %#v, err: %+v", r.Route, err)
- return err
- }
- // 2. sync apisix
if _, err := conf.Client.Cluster(cluster).Route().Update(context.TODO(), r.Route); err != nil {
log.Errorf("failed to update route %s: %s, ", *r.Name, err)
return err
}
log.Infof("update route %s, %s", *r.Name, *r.ServiceId)
} else {
- // 1. sync apisix and get id
route, err := conf.Client.Cluster(cluster).Route().Create(context.TODO(), r.Route)
if err != nil {
log.Errorf("failed to create route: %s", err.Error())
@@ -165,11 +158,6 @@ func (r *routeWorker) sync() error {
}
*r.ID = *route.ID
}
- // 2. sync memDB
- db := &db.RouteDB{Routes: []*v1.Route{r.Route}}
- if err := db.Insert(); err != nil {
- return err
- }
log.Infof("create route %s, %s", *r.Name, *r.ServiceId)
return nil
}
@@ -195,28 +183,45 @@ func SolverUpstream(upstreams []*v1.Upstream, swg ServiceWorkerGroup, wg *sync.W
}
func SolverSingleUpstream(u *v1.Upstream, swg ServiceWorkerGroup, wg *sync.WaitGroup, errorChan chan CRDStatus) {
- var errNotify error
+ var (
+ op string
+ errNotify error
+ )
defer func() {
if errNotify != nil {
errorChan <- CRDStatus{Id: "", Status: "failure", Err: errNotify}
}
wg.Done()
}()
- op := Update
- if currentUpstream, err := apisix.FindCurrentUpstream(*u.Group, *u.Name, *u.FullName); err != nil {
- log.Errorf("solver upstream failed, find upstream from APISIX failed, upstream: %+v, err: %+v", u, err)
+ var cluster string
+ if u.Group != nil {
+ cluster = *u.Group
+ }
+ if currentUpstream, err := conf.Client.Cluster(cluster).Upstream().Get(context.TODO(), *u.FullName); err != nil && err != cache.ErrNotFound {
+ log.Errorf("failed to find upstream %s: %s", *u.FullName, err)
errNotify = err
return
} else {
paddingUpstream(u, currentUpstream)
- // diff
- hasDiff, err := utils.HasDiff(u, currentUpstream)
- if err != nil {
- errNotify = err
- return
- }
- if hasDiff {
- if *u.ID != strconv.Itoa(0) {
+ if currentUpstream == nil {
+ if u.FromKind != nil && *u.FromKind == WatchFromKind {
+ // We don't have a pre-defined upstream and the current upstream updating from
+ // endpoints.
+ return
+ }
+ op = Create
+ if _, err := conf.Client.Cluster(cluster).Upstream().Create(context.TODO(), u); err != nil {
+ log.Errorf("failed to create upstream %s: %s", *u.FullName, err)
+ return
+ }
+ } else {
+ // diff
+ hasDiff, err := utils.HasDiff(u, currentUpstream)
+ if err != nil {
+ errNotify = err
+ return
+ }
+ if hasDiff {
op = Update
// 0.field check
needToUpdate := true
@@ -227,64 +232,17 @@ func SolverSingleUpstream(u *v1.Upstream, swg ServiceWorkerGroup, wg *sync.WaitG
needToUpdate = false
}
}
- if needToUpdate {
- // 1.sync memDB
- upstreamDB := &db.UpstreamDB{Upstreams: []*v1.Upstream{u}}
- if err := upstreamDB.UpdateUpstreams(); err != nil {
- log.Errorf("solver upstream failed, update upstream to local db failed, err: %s", err.Error())
- errNotify = err
- return
- }
-
- // 2.sync apisix
- var cluster string
- if u.Group != nil {
- cluster = *u.Group
- }
- if _, err = conf.Client.Cluster(cluster).Upstream().Update(context.TODO(), u); err != nil {
- glog.Errorf("solver upstream failed, update upstream to etcd failed, err: %+v", err)
- return
- }
- }
- // if fromKind == WatchFromKind
- if u.FromKind != nil && *u.FromKind == WatchFromKind {
- // 1.update nodes
- if err = apisix.PatchNodes(u, u.Nodes); err != nil {
- log.Errorf("solver upstream failed, patch node info to etcd failed, err: %+v", err)
- errNotify = err
- return
- }
- // 2. sync memDB
- us := []*v1.Upstream{u}
- if !needToUpdate {
+ if needToUpdate || (u.FromKind != nil && *u.FromKind == WatchFromKind) {
+ if u.FromKind != nil && *u.FromKind == WatchFromKind {
currentUpstream.Nodes = u.Nodes
- us = []*v1.Upstream{currentUpstream}
+ } else { // due to CRD update
+ currentUpstream = u
}
- upstreamDB := &db.UpstreamDB{Upstreams: us}
- if err := upstreamDB.UpdateUpstreams(); err != nil {
- log.Errorf("solver upstream failed, update upstream to local db failed, err: %s", err.Error())
- errNotify = err
+ if _, err = conf.Client.Cluster(cluster).Upstream().Update(context.TODO(), currentUpstream); err != nil {
+ log.Errorf("failed to update upstream %s: %s", *u.FullName, err)
return
}
}
- } else {
- op = Create
- // 1.sync apisix and get response
- var cluster string
- if u.Group != nil {
- cluster = *u.Group
- }
- ups, err := conf.Client.Cluster(cluster).Upstream().Create(context.TODO(), u)
- if err != nil {
- log.Errorf("failed to create upstream: %s", err)
- return
- }
-
- *u.ID = *ups.ID
- // 2.sync memDB
- //apisix.InsertUpstreams([]*v1.Upstream{u})
- upstreamDB := &db.UpstreamDB{Upstreams: []*v1.Upstream{u}}
- upstreamDB.InsertUpstreams()
}
}
}
diff --git a/pkg/seven/state/service_worker.go b/pkg/seven/state/service_worker.go
index d1fcce3..87b654f 100644
--- a/pkg/seven/state/service_worker.go
+++ b/pkg/seven/state/service_worker.go
@@ -20,9 +20,7 @@ import (
"sync"
"github.com/api7/ingress-controller/pkg/log"
- "github.com/api7/ingress-controller/pkg/seven/apisix"
"github.com/api7/ingress-controller/pkg/seven/conf"
- "github.com/api7/ingress-controller/pkg/seven/db"
"github.com/api7/ingress-controller/pkg/seven/utils"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
)
@@ -89,7 +87,11 @@ func SolverSingleService(svc *v1.Service, rwg RouteWorkerGroup, wg *sync.WaitGro
op := Update
// padding
- currentService, _ := apisix.FindCurrentService(*svc.Group, *svc.Name, *svc.FullName)
+ var cluster string
+ if svc.Group != nil {
+ cluster = *svc.Group
+ }
+ currentService, _ := conf.Client.Cluster(cluster).Service().Get(context.TODO(), *svc.FullName)
paddingService(svc, currentService)
// diff
hasDiff, err := utils.HasDiff(svc, currentService)
@@ -98,14 +100,9 @@ func SolverSingleService(svc *v1.Service, rwg RouteWorkerGroup, wg *sync.WaitGro
errNotify = err
return
}
- var cluster string
- if svc.Group != nil {
- cluster = *svc.Group
- }
if hasDiff {
if *svc.ID == strconv.Itoa(0) {
op = Create
- // 1. sync apisix and get id
if s, err := conf.Client.Cluster(cluster).Service().Create(context.TODO(), svc); err != nil {
log.Errorf("failed to create service: %s", err)
errNotify = err
@@ -113,12 +110,6 @@ func SolverSingleService(svc *v1.Service, rwg RouteWorkerGroup, wg *sync.WaitGro
} else {
*svc.ID = *s.ID
}
- // 2. sync memDB
- db := &db.ServiceDB{Services: []*v1.Service{svc}}
- if err := db.Insert(); err != nil {
- errNotify = err
- return
- }
log.Infof("create service %s, %s", *svc.Name, *svc.UpstreamId)
} else {
op = Update
@@ -131,14 +122,6 @@ func SolverSingleService(svc *v1.Service, rwg RouteWorkerGroup, wg *sync.WaitGro
}
}
if needToUpdate {
- // 1. sync memDB
- db := db.ServiceDB{Services: []*v1.Service{svc}}
- if err := db.UpdateService(); err != nil {
- log.Errorf("failed to update service to mem db: %s", err)
- errNotify = err
- return
- }
- // 2. sync apisix
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)
diff --git a/pkg/seven/state/solver.go b/pkg/seven/state/solver.go
index 02a1d4b..a278aaa 100644
--- a/pkg/seven/state/solver.go
+++ b/pkg/seven/state/solver.go
@@ -17,14 +17,14 @@ package state
import (
"context"
"errors"
- "fmt"
"sync"
"time"
+ "go.uber.org/multierr"
+
+ "github.com/api7/ingress-controller/pkg/apisix/cache"
"github.com/api7/ingress-controller/pkg/log"
"github.com/api7/ingress-controller/pkg/seven/conf"
- "github.com/api7/ingress-controller/pkg/seven/db"
- "github.com/api7/ingress-controller/pkg/seven/utils"
v1 "github.com/api7/ingress-controller/pkg/types/apisix/v1"
)
@@ -57,92 +57,58 @@ func (s *ApisixCombination) Solver() (string, error) {
func (s *ApisixCombination) Remove() error {
// services
for _, svc := range s.Services {
- if err := RemoveService(svc); err != nil {
- return err
+ var cluster string
+ if svc.Group != nil {
+ cluster = *svc.Group
}
- }
-
- // upstreams
- for _, up := range s.Upstreams {
- if err := RemoveUpstream(up); err != nil {
- return err
- }
- }
- return nil
-}
-
-func RemoveService(svc *v1.Service) error {
- // find svc ID
- serviceRequest := db.ServiceRequest{FullName: *svc.FullName}
- if service, err := serviceRequest.FindByName(); err != nil {
- // if not found, keep going on.
- if errors.Is(err, utils.ErrNotFound) {
- return nil
- }
- return err
- } else {
- svc.ID = service.ID
- }
- // find ref route
- routeRequest := db.RouteRequest{ServiceId: *svc.ID}
- if route, err := routeRequest.ExistByServiceId(); err != nil {
- if !errors.Is(err, utils.ErrNotFound) {
- // except ErrNotFound, need to retry
- return err
- } else {
- // do delete svc
- var cluster string
- if svc.Group != nil {
- cluster = *svc.Group
- }
- if err := conf.Client.Cluster(cluster).Service().Delete(context.TODO(), svc); err != nil {
- log.Errorf("failed to delete svc %s from APISIX: %s", *svc.FullName, err)
+ svcInCache, err := conf.Client.Cluster(cluster).Service().Get(context.TODO(), *svc.FullName)
+ if err != nil {
+ if err == cache.ErrNotFound {
+ log.Errorf("failed to remove service %s: %s", *svc.FullName, err)
+ continue
+ } else {
return err
+ }
+ }
+ paddingService(svc, svcInCache)
+ err = conf.Client.Cluster(cluster).Service().Delete(context.TODO(), svc)
+ if err != nil {
+ if err == cache.ErrNotFound {
+ log.Errorf("failed to remove service %s: %s", *svc.FullName, err)
+ } else if err == cache.ErrStillInUse {
+ log.Warnf("failed to remove service %s: %s", *svc.FullName, err)
} else {
- db := db.ServiceDB{Services: []*v1.Service{svc}}
- db.DeleteService()
- return nil
+ return err
}
}
- } else {
- return fmt.Errorf("svc %s is still referenced by route %s", *svc.FullName, *route.FullName)
}
-}
-func RemoveUpstream(up *v1.Upstream) error {
- upstreamRequest := db.UpstreamRequest{FullName: *up.FullName}
- if upstream, err := upstreamRequest.FindByName(); err != nil {
- // if not found, keep going on.
- if errors.Is(err, utils.ErrNotFound) {
- return nil
+ // upstreams
+ for _, ups := range s.Upstreams {
+ var cluster string
+ if ups.Group != nil {
+ cluster = *ups.Group
}
- return err
- } else {
- up.ID = upstream.ID
- }
- serviceRequest := db.ServiceRequest{UpstreamId: *up.ID}
- if svc, err := serviceRequest.ExistByUpstreamId(); err != nil {
- if !errors.Is(err, utils.ErrNotFound) {
- // except ErrNotFound, need to retry
- return err
- } else {
- // do delete upstream
- var cluster string
- if up.Group != nil {
- cluster = *up.Group
- }
- if err := conf.Client.Cluster(cluster).Upstream().Delete(context.TODO(), up); err != nil {
- log.Errorf("failed to delete upstream %s from APISIX: %s", *up.FullName, err)
- return err
+ upsInCache, err := conf.Client.Cluster(cluster).Upstream().Get(context.TODO(), *ups.FullName)
+ if err != nil {
+ if err == cache.ErrNotFound {
+ log.Errorf("failed to remove service %s: %s", *ups.FullName, err)
+ continue
} else {
- db := db.UpstreamDB{Upstreams: []*v1.Upstream{up}}
- db.DeleteUpstream()
- return nil
+ return err
}
}
- } else {
- return fmt.Errorf("upstream %s is still referenced by service %s", *up.FullName, *svc.FullName)
+ paddingUpstream(ups, upsInCache)
+ err = conf.Client.Cluster(cluster).Upstream().Delete(context.TODO(), ups)
+ if err == cache.ErrNotFound {
+ log.Errorf("failed to remove upstream %s: %s", *ups.FullName, err)
+ } else if err == cache.ErrStillInUse {
+ log.Warnf("failed to remove upstream %s: %s", *ups.FullName, err)
+ } else {
+ return err
+ }
}
+ return nil
}
func waitTimeout(ctx context.Context, wg *sync.WaitGroup, resultChan chan CRDStatus) {
@@ -213,6 +179,7 @@ type ServiceQueueObj struct {
// Sync remove from apisix
func (rc *RouteCompare) Sync() error {
+ var merr error
for _, old := range rc.OldRoutes {
needToDel := true
for _, nr := range rc.NewRoutes {
@@ -222,29 +189,28 @@ func (rc *RouteCompare) Sync() error {
}
}
if needToDel {
- fullName := *old.Name
- if *old.Group != "" {
- fullName = *old.Group + "_" + *old.Name
+ var cluster string
+ if old.Group != nil {
+ cluster = *old.Group
}
- request := db.RouteRequest{Name: *old.Name, FullName: fullName}
- if route, err := request.FindByName(); err != nil {
- log.Errorf("failed to find route %s from memory DB: %s", *old.Name, err)
- } else {
- var cluster string
- if route.Group != nil {
- cluster = *route.Group
- }
- if err := conf.Client.Cluster(cluster).Route().Delete(context.TODO(), route); err != nil {
- log.Errorf("failed to delete route %s from APISIX: %s", *route.Name, err)
- } else {
- db := db.RouteDB{Routes: []*v1.Route{route}}
- db.DeleteRoute()
+ // old should inject the ID.
+ route, err := conf.Client.Cluster(cluster).Route().Get(context.TODO(), *old.FullName)
+ if err != nil {
+ if err != cache.ErrNotFound {
+ merr = multierr.Append(merr, err)
}
+ continue
+ }
+
+ paddingRoute(old, route)
+ if err := conf.Client.Cluster(cluster).Route().Delete(context.TODO(), old); err != nil {
+ log.Errorf("failed to delete route %s from APISIX: %s", *old.Name, err)
+ merr = multierr.Append(merr, err)
}
}
}
- return nil
+ return merr
}
func SyncSsl(ssl *v1.Ssl, method string) error {
@@ -257,9 +223,11 @@ func SyncSsl(ssl *v1.Ssl, method string) error {
_, err := conf.Client.Cluster(cluster).SSL().Create(context.TODO(), ssl)
return err
case Update:
+ // FIXME we don't know the full name of SSL.
_, err := conf.Client.Cluster(cluster).SSL().Update(context.TODO(), ssl)
return err
case Delete:
+ // FIXME we don't know the full name of SSL.
return conf.Client.Cluster(cluster).SSL().Delete(context.TODO(), ssl)
}
return nil
diff --git a/pkg/seven/utils/diff.go b/pkg/seven/utils/diff.go
index 10a74ea..5a48256 100644
--- a/pkg/seven/utils/diff.go
+++ b/pkg/seven/utils/diff.go
@@ -16,6 +16,7 @@ package utils
import (
"encoding/json"
+
"github.com/golang/glog"
"github.com/yudai/gojsondiff"
)
diff --git a/test/e2e/scaffold/crd.go b/test/e2e/scaffold/crd.go
index b955270..88022a5 100644
--- a/test/e2e/scaffold/crd.go
+++ b/test/e2e/scaffold/crd.go
@@ -192,7 +192,11 @@ func (s *Scaffold) ListApisixUpstreams() ([]*v1.Upstream, error) {
Host: host,
Path: "/apisix/admin",
}
- cli, err := apisix.NewForOptions(&apisix.ClusterOptions{
+ cli, err := apisix.NewClient()
+ if err != nil {
+ return nil, err
+ }
+ err = cli.AddCluster(&apisix.ClusterOptions{
BaseURL: u.String(),
})
if err != nil {