feat: implement schema API (#601)

.golangci.yml

@@ -23,3 +23,11 @@ linters-settings:
   govet:
     disable:
     - unsafeptr
+  goimports:
+    local-prefixes: github.com/apache/apisix-ingress-controller
+
+linters:
+  enable:
+    - goimports
+    - govet
+    - gofmt

pkg/apisix/apisix.go

@@ -12,6 +12,7 @@
 // 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 (
@@ -26,9 +27,9 @@ type APISIX interface {
 	// Cluster specifies the target cluster to talk.
 	Cluster(string) Cluster
 	// AddCluster adds a new cluster.
-	AddCluster(*ClusterOptions) error
+	AddCluster(context.Context, *ClusterOptions) error
 	// UpdateCluster updates an existing cluster.
-	UpdateCluster(*ClusterOptions) error
+	UpdateCluster(context.Context, *ClusterOptions) error
 	// ListClusters lists all APISIX clusters.
 	ListClusters() []Cluster
 }
@@ -54,6 +55,10 @@ type Cluster interface {
 	Consumer() Consumer
 	// HealthCheck checks apisix cluster health in realtime.
 	HealthCheck(context.Context) error
+	// Plugin returns a Plugin interface that can operate Plugin resources.
+	Plugin() Plugin
+	// Schema returns a Schema interface that can fetch schema of APISIX objects.
+	Schema() Schema
 }
 
 // Route is the specific client interface to take over the create, update,
@@ -106,7 +111,7 @@ type GlobalRule interface {
 	Update(context.Context, *v1.GlobalRule) (*v1.GlobalRule, error)
 }
 
-// Consumer it the specific client interface to take over the create, update,
+// Consumer is the specific client interface to take over the create, update,
 // list and delete for APISIX Consumer resource.
 type Consumer interface {
 	Get(context.Context, string) (*v1.Consumer, error)
@@ -116,6 +121,16 @@ type Consumer interface {
 	Update(context.Context, *v1.Consumer) (*v1.Consumer, error)
 }
 
+// Plugin is the specific client interface to fetch APISIX Plugin resource.
+type Plugin interface {
+	List(context.Context) ([]string, error)
+}
+
+// Schema is the specific client interface to fetch the schema of APISIX objects.
+type Schema interface {
+	GetPluginSchema(context.Context, string) (*v1.Schema, error)
+}
+
 type apisix struct {
 	mu                 sync.RWMutex
 	nonExistentCluster Cluster
@@ -154,29 +169,29 @@ func (c *apisix) ListClusters() []Cluster {
 }
 
 // AddCluster implements APISIX.AddCluster method.
-func (c *apisix) AddCluster(co *ClusterOptions) error {
+func (c *apisix) AddCluster(ctx context.Context, co *ClusterOptions) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	_, ok := c.clusters[co.Name]
 	if ok {
 		return ErrDuplicatedCluster
 	}
-	cluster, err := newCluster(co)
+	cluster, err := newCluster(ctx, co)
 	if err != nil {
 		return err
 	}
 	c.clusters[co.Name] = cluster
 	return nil
 }
 
-func (c *apisix) UpdateCluster(co *ClusterOptions) error {
+func (c *apisix) UpdateCluster(ctx context.Context, co *ClusterOptions) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if _, ok := c.clusters[co.Name]; !ok {
 		return ErrClusterNotExist
 	}
 
-	cluster, err := newCluster(co)
+	cluster, err := newCluster(ctx, co)
 	if err != nil {
 		return err
 	}

pkg/apisix/cache/cache.go

@@ -35,6 +35,8 @@ type Cache interface {
 	InsertGlobalRule(*v1.GlobalRule) error
 	// InsertConsumer adds or updates consumer to cache.
 	InsertConsumer(*v1.Consumer) error
+	// InsertSchema adds or updates schema to cache.
+	InsertSchema(*v1.Schema) error
 
 	// GetRoute finds the route from cache according to the primary index (id).
 	GetRoute(string) (*v1.Route, error)
@@ -48,6 +50,8 @@ type Cache interface {
 	GetGlobalRule(string) (*v1.GlobalRule, error)
 	// GetConsumer finds the consumer from cache according to the primary index (id).
 	GetConsumer(string) (*v1.Consumer, error)
+	// GetSchema finds the scheme from cache according to the primary index (id).
+	GetSchema(string) (*v1.Schema, error)
 
 	// ListRoutes lists all routes in cache.
 	ListRoutes() ([]*v1.Route, error)
@@ -61,6 +65,8 @@ type Cache interface {
 	ListGlobalRules() ([]*v1.GlobalRule, error)
 	// ListConsumers lists all consumer objects in cache.
 	ListConsumers() ([]*v1.Consumer, error)
+	// ListSchema lists all schema in cache.
+	ListSchema() ([]*v1.Schema, error)
 
 	// DeleteRoute deletes the specified route in cache.
 	DeleteRoute(*v1.Route) error
@@ -74,4 +80,6 @@ type Cache interface {
 	DeleteGlobalRule(*v1.GlobalRule) error
 	// DeleteConsumer deletes the specified consumer in cache.
 	DeleteConsumer(*v1.Consumer) error
+	// DeleteSchema deletes the specified schema in cache.
+	DeleteSchema(*v1.Schema) error
 }

pkg/apisix/cache/memdb.go

@@ -70,6 +70,10 @@ func (c *dbCache) InsertConsumer(consumer *v1.Consumer) error {
 	return c.insert("consumer", consumer.DeepCopy())
 }
 
+func (c *dbCache) InsertSchema(schema *v1.Schema) error {
+	return c.insert("schema", schema.DeepCopy())
+}
+
 func (c *dbCache) insert(table string, obj interface{}) error {
 	txn := c.db.Txn(true)
 	defer txn.Abort()
@@ -128,6 +132,14 @@ func (c *dbCache) GetConsumer(username string) (*v1.Consumer, error) {
 	return obj.(*v1.Consumer).DeepCopy(), nil
 }
 
+func (c *dbCache) GetSchema(name string) (*v1.Schema, error) {
+	obj, err := c.get("schema", name)
+	if err != nil {
+		return nil, err
+	}
+	return obj.(*v1.Schema).DeepCopy(), nil
+}
+
 func (c *dbCache) get(table, id string) (interface{}, error) {
 	txn := c.db.Txn(false)
 	defer txn.Abort()
@@ -216,6 +228,18 @@ func (c *dbCache) ListConsumers() ([]*v1.Consumer, error) {
 	return consumers, nil
 }
 
+func (c *dbCache) ListSchema() ([]*v1.Schema, error) {
+	raws, err := c.list("schema")
+	if err != nil {
+		return nil, err
+	}
+	schemaList := make([]*v1.Schema, 0, len(raws))
+	for _, raw := range raws {
+		schemaList = append(schemaList, raw.(*v1.Schema).DeepCopy())
+	}
+	return schemaList, nil
+}
+
 func (c *dbCache) list(table string) ([]interface{}, error) {
 	txn := c.db.Txn(false)
 	defer txn.Abort()
@@ -257,6 +281,10 @@ func (c *dbCache) DeleteConsumer(consumer *v1.Consumer) error {
 	return c.delete("consumer", consumer)
 }
 
+func (c *dbCache) DeleteSchema(schema *v1.Schema) error {
+	return c.delete("schema", schema)
+}
+
 func (c *dbCache) delete(table string, obj interface{}) error {
 	txn := c.db.Txn(true)
 	defer txn.Abort()

pkg/apisix/cache/memdb_test.go

@@ -345,3 +345,47 @@ func TestMemDBCacheConsumer(t *testing.T) {
 	}
 	assert.Error(t, ErrNotFound, c.DeleteConsumer(c4))
 }
+
+func TestMemDBCacheSchema(t *testing.T) {
+	c, err := NewMemDBCache()
+	assert.Nil(t, err, "NewMemDBCache")
+
+	s1 := &v1.Schema{
+		Name:    "plugins/p1",
+		Content: "plugin schema",
+	}
+	assert.Nil(t, c.InsertSchema(s1), "inserting schema s1")
+
+	s11, err := c.GetSchema("plugins/p1")
+	assert.Nil(t, err)
+	assert.Equal(t, s1, s11)
+
+	s2 := &v1.Schema{
+		Name: "plugins/p2",
+	}
+	s3 := &v1.Schema{
+		Name: "plugins/p3",
+	}
+	assert.Nil(t, c.InsertSchema(s2), "inserting schema s2")
+	assert.Nil(t, c.InsertSchema(s3), "inserting schema s3")
+
+	s22, err := c.GetSchema("plugins/p2")
+	assert.Nil(t, err)
+	assert.Equal(t, s2, s22)
+
+	assert.Nil(t, c.DeleteSchema(s3), "delete schema s3")
+
+	schemaList, err := c.ListSchema()
+	assert.Nil(t, err, "listing schema")
+
+	if schemaList[0].Name > schemaList[1].Name {
+		schemaList[0], schemaList[1] = schemaList[1], schemaList[0]
+	}
+	assert.Equal(t, schemaList[0], s1)
+	assert.Equal(t, schemaList[1], s2)
+
+	s4 := &v1.Schema{
+		Name: "plugins/p4",
+	}
+	assert.Error(t, ErrNotFound, c.DeleteSchema(s4))
+}

pkg/apisix/cache/schema.go

@@ -106,6 +106,16 @@ var (
 					},
 				},
 			},
+			"schema": {
+				Name: "schema",
+				Indexes: map[string]*memdb.IndexSchema{
+					"id": {
+						Name:    "id",
+						Unique:  true,
+						Indexer: &memdb.StringFieldIndex{Field: "Name"},
+					},
+				},
+			},
 		},
 	}
 )