allow setting listener access log via listenerPatch plugin (#701)

Signed-off-by: spacewander <[email protected]>

controller/internal/istio/envoyfilter.go

@@ -206,13 +206,36 @@ func GenerateRouteFilter(host *model.VirtualHost, route string, config map[strin
 	}
 }
 
-func GenerateLDSFilterViaECDS(key string, ldsName string, hasHCM bool, config map[string]interface{}) *istiov1a3.EnvoyFilter {
+func GenerateLDSFilter(key string, ldsName string, hasHCM bool, config map[string]interface{}) *istiov1a3.EnvoyFilter {
 	ef := &istiov1a3.EnvoyFilter{
 		Spec: istioapi.EnvoyFilter{},
 	}
 
-	if config[model.ECDSListenerFilter] != nil {
-		cfg, _ := config[model.ECDSListenerFilter].([]*fmModel.FilterConfig)
+	if config[model.CategoryListener] != nil {
+		cfg, _ := config[model.CategoryListener].([]*fmModel.FilterConfig)
+		for _, filter := range cfg {
+			c, _ := filter.Config.(map[string]interface{})
+			ef.Spec.ConfigPatches = append(ef.Spec.ConfigPatches,
+				&istioapi.EnvoyFilter_EnvoyConfigObjectPatch{
+					ApplyTo: istioapi.EnvoyFilter_LISTENER,
+					Match: &istioapi.EnvoyFilter_EnvoyConfigObjectMatch{
+						ObjectTypes: &istioapi.EnvoyFilter_EnvoyConfigObjectMatch_Listener{
+							Listener: &istioapi.EnvoyFilter_ListenerMatch{
+								Name: ldsName,
+							},
+						},
+					},
+					Patch: &istioapi.EnvoyFilter_Patch{
+						Operation: istioapi.EnvoyFilter_Patch_MERGE,
+						Value:     MustNewStruct(c),
+					},
+				},
+			)
+		}
+	}
+
+	if config[model.CategoryECDSListener] != nil {
+		cfg, _ := config[model.CategoryECDSListener].([]*fmModel.FilterConfig)
 		for i := len(cfg) - 1; i >= 0; i-- {
 			filter := cfg[i]
 			ecdsName := key + "-" + filter.Name
@@ -257,8 +280,8 @@ func GenerateLDSFilterViaECDS(key string, ldsName string, hasHCM bool, config ma
 		}
 	}
 
-	if config[model.ECDSNetworkFilter] != nil {
-		cfg, _ := config[model.ECDSNetworkFilter].([]*fmModel.FilterConfig)
+	if config[model.CategoryECDSNetwork] != nil {
+		cfg, _ := config[model.CategoryECDSNetwork].([]*fmModel.FilterConfig)
 		for i := len(cfg) - 1; i >= 0; i-- {
 			filter := cfg[i]
 			ecdsName := key + "-" + filter.Name
@@ -304,11 +327,11 @@ func GenerateLDSFilterViaECDS(key string, ldsName string, hasHCM bool, config ma
 	}
 
 	if hasHCM {
-		cfg := config[model.ECDSGolangFilter]
+		cfg := config[model.CategoryECDSGolang]
 		if cfg == nil {
 			cfg = map[string]interface{}{}
 		}
-		ecdsName := key + "-" + model.GolangPluginsFilter
+		ecdsName := key + "-" + model.CategoryGolangPlugins
 		ef.Spec.ConfigPatches = append(ef.Spec.ConfigPatches,
 			&istioapi.EnvoyFilter_EnvoyConfigObjectPatch{
 				ApplyTo: istioapi.EnvoyFilter_HTTP_FILTER,

controller/internal/model/model.go

@@ -47,9 +47,9 @@ type VirtualHost struct {
 }
 
 const (
-	ECDSGolangFilter   = "golang"
-	ECDSListenerFilter = "listener"
-	ECDSNetworkFilter  = "network"
-
-	GolangPluginsFilter = "golang-filter"
+	CategoryECDSGolang    = "ecds_golang"
+	CategoryECDSListener  = "ecds_listener"
+	CategoryECDSNetwork   = "ecds_network"
+	CategoryListener      = "listener"
+	CategoryGolangPlugins = "golang-filter"
 )

controller/internal/translation/final_state.go

@@ -131,7 +131,7 @@ func toFinalState(_ *Ctx, state *mergedState) (*FinalState, error) {
 				info = gateway.Policy.Info
 			}
 
-			ef := istio.GenerateLDSFilterViaECDS(key, name, gateway.Gateway.HasHCM, config)
+			ef := istio.GenerateLDSFilter(key, name, gateway.Gateway.HasHCM, config)
 			ef.SetNamespace(ns)
 			// Put all LDS level filters of the same LDS into the same EnvoyFilter.
 			efName := envoyFilterNameFromLds(name)

controller/internal/translation/merged_state.go

@@ -86,7 +86,7 @@ type PolicyKind int
 
 const (
 	PolicyKindRDS PolicyKind = iota
-	PolicyKindECDS
+	PolicyKindLDS
 )
 
 func translateFilterManagerConfigToPolicyInRDS(fmc *filtermanager.FilterManagerConfig,
@@ -173,7 +173,7 @@ func translateFilterManagerConfigToPolicyInRDS(fmc *filtermanager.FilterManagerC
 
 		golangFilterName := "htnn.filters.http.golang"
 		if ctrlcfg.EnableLDSPluginViaECDS() {
-			golangFilterName = virtualHost.ECDSResourceName + "-" + model.GolangPluginsFilter
+			golangFilterName = virtualHost.ECDSResourceName + "-" + model.CategoryGolangPlugins
 		}
 		config[golangFilterName] = map[string]interface{}{
 			"@type": "type.googleapis.com/envoy.extensions.filters.http.golang.v3alpha.ConfigsPerRoute",
@@ -196,15 +196,16 @@ func translateFilterManagerConfigToPolicyInRDS(fmc *filtermanager.FilterManagerC
 	return config
 }
 
-func translateFilterManagerConfigToPolicyInECDS(fmc *filtermanager.FilterManagerConfig, nsName *types.NamespacedName) map[string]interface{} {
+func translateFilterManagerConfigToPolicyInLDS(fmc *filtermanager.FilterManagerConfig, nsName *types.NamespacedName) map[string]interface{} {
 	config := map[string]interface{}{}
 
 	goFilterManager := &filtermanager.FilterManagerConfig{
 		Plugins: []*fmModel.FilterConfig{},
 	}
 	nativeFilters := map[string][]*fmModel.FilterConfig{
-		model.ECDSListenerFilter: {},
-		model.ECDSNetworkFilter:  {},
+		model.CategoryECDSListener: {},
+		model.CategoryECDSNetwork:  {},
+		model.CategoryListener:     {},
 	}
 
 	consumerNeeded := false
@@ -243,19 +244,28 @@ func translateFilterManagerConfigToPolicyInECDS(fmc *filtermanager.FilterManager
 				continue
 			}
 
-			url := nativePlugin.ConfigTypeURL()
 			m, ok := cfg.(map[string]interface{})
 			if !ok {
 				panic(fmt.Sprintf("unexpected type: %s", reflect.TypeOf(cfg)))
 			}
 
-			m["@type"] = url
-			plugin.Config = m
-
-			if order.Position == plugins.OrderPositionListener {
-				nativeFilters[model.ECDSListenerFilter] = append(nativeFilters[model.ECDSListenerFilter], plugin)
-			} else if order.Position == plugins.OrderPositionNetwork {
-				nativeFilters[model.ECDSNetworkFilter] = append(nativeFilters[model.ECDSNetworkFilter], plugin)
+			url := nativePlugin.ConfigTypeURL()
+			if url != "" {
+				m["@type"] = url
+				plugin.Config = m
+
+				if order.Position == plugins.OrderPositionListener {
+					nativeFilters[model.CategoryECDSListener] = append(nativeFilters[model.CategoryECDSListener], plugin)
+				} else if order.Position == plugins.OrderPositionNetwork {
+					nativeFilters[model.CategoryECDSNetwork] = append(nativeFilters[model.CategoryECDSNetwork], plugin)
+				}
+			} else {
+				plugin.Config = m
+
+				if order.Position == plugins.OrderPositionListener {
+					nativeFilters[model.CategoryListener] = append(nativeFilters[model.CategoryListener], plugin)
+				}
+				// TODO: support network filter
 			}
 		}
 	}
@@ -277,7 +287,7 @@ func translateFilterManagerConfigToPolicyInECDS(fmc *filtermanager.FilterManager
 			}
 		}
 		cfg["plugins"] = plugins
-		config[model.ECDSGolangFilter] = cfg
+		config[model.CategoryECDSGolang] = cfg
 	}
 
 	for category, filters := range nativeFilters {
@@ -325,8 +335,8 @@ func toMergedPolicy(nsName *types.NamespacedName, policies []*FilterPolicyWrappe
 	var config map[string]interface{}
 	if policyKind == PolicyKindRDS {
 		config = translateFilterManagerConfigToPolicyInRDS(fmc, nsName, virtualHost)
-	} else if policyKind == PolicyKindECDS {
-		config = translateFilterManagerConfigToPolicyInECDS(fmc, nsName)
+	} else if policyKind == PolicyKindLDS {
+		config = translateFilterManagerConfigToPolicyInLDS(fmc, nsName)
 	}
 
 	return &mergedPolicy{
@@ -384,7 +394,7 @@ func toMergedState(ctx *Ctx, state *dataPlaneState) (*FinalState, error) {
 				Gateway: gateway.Gateway,
 			}
 			if len(gateway.Policies) > 0 {
-				mg.Policy = toMergedPolicy(&gateway.Gateway.GatewaySection.NsName, gateway.Policies, PolicyKindECDS, nil)
+				mg.Policy = toMergedPolicy(&gateway.Gateway.GatewaySection.NsName, gateway.Policies, PolicyKindLDS, nil)
 			}
 
 			mergedGateways[name] = mg

controller/plugins/listenerpatch/config.go

@@ -0,0 +1,32 @@
+// Copyright The HTNN Authors.
+//
+// Licensed 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 listenerpatch
+
+import (
+	"mosn.io/htnn/api/pkg/plugins"
+	"mosn.io/htnn/types/plugins/listenerpatch"
+)
+
+func init() {
+	plugins.RegisterPlugin(listenerpatch.Name, &plugin{})
+}
+
+type plugin struct {
+	listenerpatch.Plugin
+}
+
+func (p *plugin) ConfigTypeURL() string {
+	return ""
+}

controller/plugins/plugins.go

@@ -20,6 +20,7 @@ import (
 	_ "mosn.io/htnn/controller/plugins/cors"
 	_ "mosn.io/htnn/controller/plugins/extproc"
 	_ "mosn.io/htnn/controller/plugins/fault"
+	_ "mosn.io/htnn/controller/plugins/listenerpatch"
 	_ "mosn.io/htnn/controller/plugins/localratelimit"
 	_ "mosn.io/htnn/controller/plugins/lua"
 	_ "mosn.io/htnn/controller/plugins/networkrbac"

controller/plugins/testdata/network/listener_patch.in.yml

@@ -0,0 +1,21 @@
+apiVersion: htnn.mosn.io/v1
+kind: FilterPolicy
+metadata:
+  name: policy
+  namespace: default
+spec:
+  targetRef:
+    group: networking.istio.io
+    kind: Gateway
+    name: default
+  filters:
+    listenerPatch:
+      config:
+        accessLog:
+        - name: envoy.access_loggers.file
+          typedConfig:
+            "@type": type.googleapis.com/envoy.extensions.access_loggers.file.v3.FileAccessLog
+            path: /home/logs/access.log
+            logFormat:
+              textFormatSource:
+                inlineString: "%START_TIME%,%DOWNSTREAM_REMOTE_ADDRESS_WITHOUT_PORT%"

controller/plugins/testdata/network/listener_patch.out.yml

@@ -0,0 +1,22 @@
+- metadata:
+    creationTimestamp: null
+    name: htnn-lds-0.0.0.0-18000
+    namespace: default
+  spec:
+    configPatches:
+    - applyTo: LISTENER
+      match:
+        listener:
+          name: 0.0.0.0_18000
+      patch:
+        operation: MERGE
+        value:
+          accessLog:
+          - name: envoy.access_loggers.file
+            typedConfig:
+              '@type': type.googleapis.com/envoy.extensions.access_loggers.file.v3.FileAccessLog
+              logFormat:
+                textFormatSource:
+                  inlineString: '%START_TIME%,%DOWNSTREAM_REMOTE_ADDRESS_WITHOUT_PORT%'
+              path: /home/logs/access.log
+  status: {}

e2e/e2e_test.go

@@ -33,10 +33,12 @@ func TestE2E(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Error loading Kubernetes config: %v", err)
 	}
+
 	client, err := client.New(cfg, client.Options{})
 	if err != nil {
 		t.Fatalf("Error initializing Kubernetes client: %v", err)
 	}
+
 	clientset, err := kubernetes.NewForConfig(cfg)
 	if err != nil {
 		t.Fatalf("Error initializing Kubernetes REST client: %v", err)

e2e/pkg/suite/suite.go

@@ -31,6 +31,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/gateway-api/conformance/utils/roundtripper"
@@ -250,3 +252,30 @@ func (suite *Suite) Capture(resp *http.Response) (*roundtripper.CapturedRequest,
 
 	return cReq, cRes, nil
 }
+
+func (suite *Suite) GetLog(namespace string, prefix string) ([]byte, error) {
+	ctx := context.Background()
+	clientset := suite.Opt.Clientset
+
+	podName := ""
+	pods, err := clientset.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, pod := range pods.Items {
+		if strings.HasPrefix(pod.Name, prefix) {
+			podName = pod.Name
+			break
+		}
+	}
+
+	req := clientset.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{})
+	podLogs, err := req.Stream(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer podLogs.Close()
+
+	return io.ReadAll(podLogs)
+}

e2e/tests/listener_patch.go

@@ -0,0 +1,51 @@
+// Copyright The HTNN Authors.
+//
+// Licensed 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 tests
+
+import (
+	"bytes"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"mosn.io/htnn/e2e/pkg/k8s"
+	"mosn.io/htnn/e2e/pkg/suite"
+)
+
+func init() {
+	suite.Register(suite.Test{
+		Run: func(t *testing.T, suite *suite.Suite) {
+			hdr := http.Header{}
+			hdr.Add("Connection", "close")
+			rsp, err := suite.Get("/echo", hdr)
+			require.NoError(t, err)
+			require.Equal(t, 200, rsp.StatusCode)
+
+			time.Sleep(100 * time.Millisecond)
+			require.Eventually(t, func() bool {
+				namespace := k8s.DefaultNamespace
+
+				b, err := suite.GetLog(namespace, "default-istio-")
+				if err != nil {
+					t.Logf("unexpected error %v", err)
+					return false
+				}
+				return bytes.Contains(b, []byte("added access log: 127.0.0.1:10000"))
+			}, 10*time.Second, 100*time.Millisecond)
+		},
+	})
+}