feat: support plugin_metadata of apisix (#1369)

cmd/ingress/ingress.go

@@ -187,6 +187,7 @@ For example, no available LB exists in the bare metal environment.`)
 	cmd.PersistentFlags().StringVar(&cfg.APISIX.DefaultClusterAdminKey, "default-apisix-cluster-admin-key", "", "admin key used for the authorization of admin api / manager api for the default APISIX cluster")
 	cmd.PersistentFlags().StringVar(&cfg.APISIX.DefaultClusterName, "default-apisix-cluster-name", "default", "name of the default apisix cluster")
 	cmd.PersistentFlags().DurationVar(&cfg.ApisixResourceSyncInterval.Duration, "apisix-resource-sync-interval", 300*time.Second, "interval between syncs in seconds. Default value is 300s.")
+	cmd.PersistentFlags().StringVar(&cfg.PluginMetadataConfigMap, "plugin-metadata-cm", "plugin-metadata-config-map", "ConfigMap name of plugin metadata.")
 
 	if err := cmd.PersistentFlags().MarkDeprecated("app-namespace", "use namespace-selector instead"); err != nil {
 		dief("failed to mark `app-namespace` as deprecated: %s", err)

conf/config-default.yaml

@@ -84,6 +84,7 @@ kubernetes:
                                        # Before we announce support for it to reach Beta level or GA.
   api_version: apisix.apache.org/v2    # the default value of API version is "apisix.apache.org/v2", support "apisix.apache.org/v2beta3" and "apisix.apache.org/v2".
 
+  plugin_metadata_cm: plugin-metadata-config-map
 # APISIX related configurations.
 apisix:
   default_cluster_base_url: "http://127.0.0.1:9080/apisix/admin" # The base url of admin api / manager api

conf/plugin-metadata-config-map.yaml

@@ -0,0 +1,39 @@
+# 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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: plugin-metadata-config-map
+data:
+  config.yaml: |
+    - cluster: default
+      plugins:
+      - name : http-logger
+        metadata: 
+          log_format:
+            host: "$host"
+            client_ip: "$remote_addr"
+      - name: kafka-logger
+        metadata:
+          log_format:
+            host: "$host"
+      - name: datadog
+        metadata:
+          host: "DogStatsD.server.domain"
+          port: 8125
+          namespace: "apisix" 
+    - cluster: sddsass
+    - cluster: Xxxxx

pkg/apisix/apisix.go

@@ -63,6 +63,8 @@ type Cluster interface {
 	PluginConfig() PluginConfig
 	// Schema returns a Schema interface that can fetch schema of APISIX objects.
 	Schema() Schema
+
+	PluginMetadata() PluginMetadata
 	// UpstreamServiceRelation returns a UpstreamServiceRelation interface that can fetch UpstreamServiceRelation of APISIX objects.
 	UpstreamServiceRelation() UpstreamServiceRelation
 }
@@ -152,6 +154,13 @@ type PluginConfig interface {
 	Update(context.Context, *v1.PluginConfig) (*v1.PluginConfig, error)
 }
 
+type PluginMetadata interface {
+	Get(context.Context, string) (*v1.PluginMetadata, error)
+	List(context.Context) ([]*v1.PluginMetadata, error)
+	Delete(context.Context, *v1.PluginMetadata) error
+	Update(context.Context, *v1.PluginMetadata) (*v1.PluginMetadata, error)
+}
+
 type UpstreamServiceRelation interface {
 	// Get relation based on namespace+"_"+service.name
 	Get(context.Context, string) (*v1.UpstreamServiceRelation, error)

pkg/apisix/cluster.go

@@ -16,6 +16,7 @@
 package apisix
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -104,6 +105,7 @@ type cluster struct {
 	pluginConfig            PluginConfig
 	metricsCollector        metrics.Collector
 	upstreamServiceRelation UpstreamServiceRelation
+	pluginMetadata          PluginMetadata
 }
 
 func newCluster(ctx context.Context, o *ClusterOptions) (Cluster, error) {
@@ -146,6 +148,7 @@ func newCluster(ctx context.Context, o *ClusterOptions) (Cluster, error) {
 	c.schema = newSchemaClient(c)
 	c.pluginConfig = newPluginConfigClient(c)
 	c.upstreamServiceRelation = newUpstreamServiceRelation(c)
+	c.pluginMetadata = newPluginMetadataClient(c)
 
 	c.cache, err = cache.NewMemDBCache()
 	if err != nil {
@@ -462,6 +465,10 @@ func (c *cluster) Schema() Schema {
 	return c.schema
 }
 
+func (c *cluster) PluginMetadata() PluginMetadata {
+	return c.pluginMetadata
+}
+
 func (c *cluster) UpstreamServiceRelation() UpstreamServiceRelation {
 	return c.upstreamServiceRelation
 }
@@ -530,6 +537,11 @@ func (c *cluster) isFunctionDisabled(body string) bool {
 }
 
 func (c *cluster) getResource(ctx context.Context, url, resource string) (*getResponse, error) {
+	log.Debugw("get resource in cluster",
+		zap.String("cluster_name", c.name),
+		zap.String("name", resource),
+		zap.String("url", url),
+	)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, err
@@ -567,6 +579,11 @@ func (c *cluster) getResource(ctx context.Context, url, resource string) (*getRe
 }
 
 func (c *cluster) listResource(ctx context.Context, url, resource string) (*listResponse, error) {
+	log.Debugw("list resource in cluster",
+		zap.String("cluster_name", c.name),
+		zap.String("name", resource),
+		zap.String("url", url),
+	)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, err
@@ -599,8 +616,14 @@ func (c *cluster) listResource(ctx context.Context, url, resource string) (*list
 	return &list, nil
 }
 
-func (c *cluster) createResource(ctx context.Context, url, resource string, body io.Reader) (*createResponse, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, body)
+func (c *cluster) createResource(ctx context.Context, url, resource string, body []byte) (*createResponse, error) {
+	log.Debugw("creating resource in cluster",
+		zap.String("cluster_name", c.name),
+		zap.String("name", resource),
+		zap.String("url", url),
+		zap.ByteString("body", body),
+	)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(body))
 	if err != nil {
 		return nil, err
 	}
@@ -632,8 +655,14 @@ func (c *cluster) createResource(ctx context.Context, url, resource string, body
 	return &cr, nil
 }
 
-func (c *cluster) updateResource(ctx context.Context, url, resource string, body io.Reader) (*updateResponse, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, body)
+func (c *cluster) updateResource(ctx context.Context, url, resource string, body []byte) (*updateResponse, error) {
+	log.Debugw("updating resource in cluster",
+		zap.String("cluster_name", c.name),
+		zap.String("name", resource),
+		zap.String("url", url),
+		zap.ByteString("body", body),
+	)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(body))
 	if err != nil {
 		return nil, err
 	}
@@ -665,6 +694,11 @@ func (c *cluster) updateResource(ctx context.Context, url, resource string, body
 }
 
 func (c *cluster) deleteResource(ctx context.Context, url, resource string) error {
+	log.Debugw("deleting resource in cluster",
+		zap.String("cluster_name", c.name),
+		zap.String("name", resource),
+		zap.String("url", url),
+	)
 	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil)
 	if err != nil {
 		return err

pkg/apisix/consumer.go

@@ -15,7 +15,6 @@
 package apisix
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 
@@ -155,7 +154,7 @@ func (r *consumerClient) Create(ctx context.Context, obj *v1.Consumer) (*v1.Cons
 
 	url := r.url + "/" + obj.Username
 	log.Debugw("creating consumer", zap.ByteString("body", data), zap.String("url", url))
-	resp, err := r.cluster.createResource(ctx, url, "consumer", bytes.NewReader(data))
+	resp, err := r.cluster.createResource(ctx, url, "consumer", data)
 	r.cluster.metricsCollector.IncrAPISIXRequest("consumer")
 	if err != nil {
 		log.Errorf("failed to create consumer: %s", err)
@@ -212,8 +211,7 @@ func (r *consumerClient) Update(ctx context.Context, obj *v1.Consumer) (*v1.Cons
 		return nil, err
 	}
 	url := r.url + "/" + obj.Username
-	log.Debugw("updating username", zap.ByteString("body", body), zap.String("url", url))
-	resp, err := r.cluster.updateResource(ctx, url, "consumer", bytes.NewReader(body))
+	resp, err := r.cluster.updateResource(ctx, url, "consumer", body)
 	r.cluster.metricsCollector.IncrAPISIXRequest("consumer")
 	if err != nil {
 		return nil, err

pkg/apisix/global_rule.go

@@ -15,7 +15,6 @@
 package apisix
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 
@@ -157,7 +156,7 @@ func (r *globalRuleClient) Create(ctx context.Context, obj *v1.GlobalRule) (*v1.
 
 	url := r.url + "/" + obj.ID
 	log.Debugw("creating global_rule", zap.ByteString("body", data), zap.String("url", url))
-	resp, err := r.cluster.createResource(ctx, url, "globalRule", bytes.NewReader(data))
+	resp, err := r.cluster.createResource(ctx, url, "globalRule", data)
 	r.cluster.metricsCollector.IncrAPISIXRequest("globalRule")
 	if err != nil {
 		log.Errorf("failed to create global_rule: %s", err)
@@ -214,8 +213,7 @@ func (r *globalRuleClient) Update(ctx context.Context, obj *v1.GlobalRule) (*v1.
 		return nil, err
 	}
 	url := r.url + "/" + obj.ID
-	log.Debugw("updating global_rule", zap.ByteString("body", body), zap.String("url", url))
-	resp, err := r.cluster.updateResource(ctx, url, "globalRule", bytes.NewReader(body))
+	resp, err := r.cluster.updateResource(ctx, url, "globalRule", body)
 	r.cluster.metricsCollector.IncrAPISIXRequest("globalRule")
 	if err != nil {
 		return nil, err

pkg/apisix/nonexistentclient.go

@@ -39,6 +39,7 @@ func newNonExistentCluster() *nonExistentCluster {
 			schema:                  &dummySchema{},
 			pluginConfig:            &dummyPluginConfig{},
 			upstreamServiceRelation: &dummyUpstreamServiceRelation{},
+			pluginMetadata:          &dummyPluginMetadata{},
 		},
 	}
 }
@@ -54,6 +55,7 @@ type embedDummyResourceImplementer struct {
 	schema                  Schema
 	pluginConfig            PluginConfig
 	upstreamServiceRelation UpstreamServiceRelation
+	pluginMetadata          PluginMetadata
 }
 
 type dummyRoute struct{}
@@ -258,6 +260,23 @@ func (f *dummyUpstreamServiceRelation) Delete(_ context.Context, _ string) error
 	return ErrClusterNotExist
 }
 
+type dummyPluginMetadata struct {
+}
+
+func (f *dummyPluginMetadata) Get(_ context.Context, _ string) (*v1.PluginMetadata, error) {
+	return nil, ErrClusterNotExist
+}
+
+func (f *dummyPluginMetadata) List(_ context.Context) ([]*v1.PluginMetadata, error) {
+	return nil, ErrClusterNotExist
+}
+func (f *dummyPluginMetadata) Delete(_ context.Context, _ *v1.PluginMetadata) error {
+	return ErrClusterNotExist
+}
+func (f *dummyPluginMetadata) Update(_ context.Context, _ *v1.PluginMetadata) (*v1.PluginMetadata, error) {
+	return nil, ErrClusterNotExist
+}
+
 func (nc *nonExistentCluster) Route() Route {
 	return nc.route
 }
@@ -293,6 +312,9 @@ func (nc *nonExistentCluster) PluginConfig() PluginConfig {
 func (nc *nonExistentCluster) Schema() Schema {
 	return nc.schema
 }
+func (nc *nonExistentCluster) PluginMetadata() PluginMetadata {
+	return nc.pluginMetadata
+}
 
 func (nc *nonExistentCluster) UpstreamServiceRelation() UpstreamServiceRelation {
 	return nc.upstreamServiceRelation