rollback to Envoy 1.29.x (#378)

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

Makefile

@@ -20,10 +20,10 @@ TARGET_SO       = libgolang.so
 PROJECT_NAME    = mosn.io/htnn
 # Both images use glibc 2.31. Ensure libc in the images match each other.
 BUILD_IMAGE     ?= golang:1.21-bullseye
-# This is the envoyproxy/envoy:contrib-dev fetched in 2024-02-08
-# Use docker inspect --format='{{index .RepoDigests 0}}' envoyproxy/envoy:contrib-dev
+# This is the envoyproxy/envoy:contrib-v1.29-latest
+# Use docker inspect --format='{{index .RepoDigests 0}}' envoyproxy/envoy:contrib-v1.29-latest
 # to get the sha256 ID
-PROXY_IMAGE     ?= envoyproxy/envoy@sha256:4ac92ea7095c787b6d1d74ea7d27a5c776ded08a0dee9b6737f2105f790fa384
+PROXY_IMAGE     ?= envoyproxy/envoy@sha256:98ed3d86ff8b86dc12ddf54b7bb67ddf5506f80769038b3e2ab7bf402730fb4d
 # We may need to use timestamp if we need to update the image in one PR
 DEV_TOOLS_IMAGE ?= ghcr.io/mosn/htnn-dev-tools:2024-03-05
 

cmd/libgolang/main.go

@@ -30,8 +30,8 @@ var (
 )
 
 func init() {
-	http.RegisterHttpFilterFactoryAndConfigParser("fm", filtermanager.FilterManagerFactory, &filtermanager.FilterManagerConfigParser{})
-	http.RegisterHttpFilterFactoryAndConfigParser("cm", consumer.ConsumerManagerFactory, &consumer.ConsumerManagerConfigParser{})
+	http.RegisterHttpFilterConfigFactoryAndParser("fm", filtermanager.FilterManagerFactory, &filtermanager.FilterManagerConfigParser{})
+	http.RegisterHttpFilterConfigFactoryAndParser("cm", consumer.ConsumerManagerFactory, &consumer.ConsumerManagerConfigParser{})
 }
 
 func main() {}

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/casbin/casbin/v2 v2.82.0
 	github.com/cncf/xds/go v0.0.0-20231109132714-523115ebc101
 	github.com/coreos/go-oidc/v3 v3.9.0
-	github.com/envoyproxy/envoy v1.29.1-0.20240208055117-b788e1a92347
+	github.com/envoyproxy/envoy v1.29.0
 	github.com/envoyproxy/go-control-plane v0.11.2-0.20231019082134-6e4589f570e1 // version used by istio 1.20
 	github.com/envoyproxy/protoc-gen-validate v1.0.2
 	github.com/go-logr/logr v1.4.1

go.sum

@@ -62,8 +62,8 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
 github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
-github.com/envoyproxy/envoy v1.29.1-0.20240208055117-b788e1a92347 h1:WzZYszFnkdPdwQXv75W7pKM0dNCisJbffwbNr4BwKKM=
-github.com/envoyproxy/envoy v1.29.1-0.20240208055117-b788e1a92347/go.mod h1:c2OGLXJVY9BaTYPiWFRz6eUNaC+3TfqFKmkWS9brdKo=
+github.com/envoyproxy/envoy v1.29.0 h1:nTylh1h+Q+FfPi1YVOMWwd+7u/Mu5UQEw3ZI1ULtcgU=
+github.com/envoyproxy/envoy v1.29.0/go.mod h1:c2OGLXJVY9BaTYPiWFRz6eUNaC+3TfqFKmkWS9brdKo=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.11.2-0.20231019082134-6e4589f570e1 h1:i/XN+pZrv2iM+Czc4o4tP6UzUJoOxjNI9gQdE1vIjd0=
 github.com/envoyproxy/go-control-plane v0.11.2-0.20231019082134-6e4589f570e1/go.mod h1:3X10o7QcAVxP4y/hnTLgkXLwuZV2DxAEh6uaYD5PoxI=

pkg/consumer/consumer.go

@@ -34,10 +34,13 @@ type consumerManager struct {
 	conf      *consumerManagerConfig
 }
 
-func ConsumerManagerFactory(c interface{}, callbacks capi.FilterCallbackHandler) capi.StreamFilter {
-	return &consumerManager{
-		callbacks: callbacks,
-		conf:      c.(*consumerManagerConfig),
+func ConsumerManagerFactory(c interface{}) capi.StreamFilterFactory {
+	conf := c.(*consumerManagerConfig)
+	return func(callbacks capi.FilterCallbackHandler) capi.StreamFilter {
+		return &consumerManager{
+			callbacks: callbacks,
+			conf:      conf,
+		}
 	}
 }
 

pkg/filtermanager/api/api.go

@@ -159,8 +159,6 @@ type StreamInfo interface {
 	FilterState() FilterState
 	// VirtualClusterName returns the name of the virtual cluster which got matched
 	VirtualClusterName() (string, bool)
-	// WorkerID returns the ID of the Envoy worker thread
-	WorkerID() uint32
 
 	// Methods added by HTNN
 

pkg/filtermanager/filtermanager.go

@@ -378,63 +378,65 @@ func newSkipMethodsMap() map[string]bool {
 	}
 }
 
-func FilterManagerFactory(c interface{}, cb capi.FilterCallbackHandler) capi.StreamFilter {
+func FilterManagerFactory(c interface{}) capi.StreamFilterFactory {
 	conf := c.(*filterManagerConfig)
 	parsedConfig := conf.parsed
 
-	fm := conf.pool.Get().(*filterManager)
-	fm.callbacks.FilterCallbackHandler = cb
+	return func(cb capi.FilterCallbackHandler) capi.StreamFilter {
+		fm := conf.pool.Get().(*filterManager)
+		fm.callbacks.FilterCallbackHandler = cb
 
-	canSkipMethod := fm.canSkipMethod
-	if canSkipMethod == nil {
-		canSkipMethod = newSkipMethodsMap()
-	}
+		canSkipMethod := fm.canSkipMethod
+		if canSkipMethod == nil {
+			canSkipMethod = newSkipMethodsMap()
+		}
 
-	filters := make([]*model.FilterWrapper, len(parsedConfig))
-	for i, fc := range parsedConfig {
-		factory := fc.Factory
-		config := fc.ParsedConfig
-		f := factory(config, fm.callbacks)
-		// Technically, the factory might create different f for different calls. We don't support this edge case for now.
-		if fm.canSkipMethod == nil {
-			for meth := range canSkipMethod {
-				overridden, err := reflectx.IsMethodOverridden(f, meth)
-				if err != nil {
-					api.LogErrorf("failed to check method %s in filter: %v", meth, err)
-					// canSkipMethod[meth] will be false
+		filters := make([]*model.FilterWrapper, len(parsedConfig))
+		for i, fc := range parsedConfig {
+			factory := fc.Factory
+			config := fc.ParsedConfig
+			f := factory(config, fm.callbacks)
+			// Technically, the factory might create different f for different calls. We don't support this edge case for now.
+			if fm.canSkipMethod == nil {
+				for meth := range canSkipMethod {
+					overridden, err := reflectx.IsMethodOverridden(f, meth)
+					if err != nil {
+						api.LogErrorf("failed to check method %s in filter: %v", meth, err)
+						// canSkipMethod[meth] will be false
+					}
+					canSkipMethod[meth] = canSkipMethod[meth] && !overridden
 				}
-				canSkipMethod[meth] = canSkipMethod[meth] && !overridden
+				// Do we need to check if the correct method is defined? For example, the DecodeRequest
+				// requires DecodeHeaders defined. Currently, we just documentate it. Per request check
+				// is expensive and not necessary in most of time.
 			}
-			// Do we need to check if the correct method is defined? For example, the DecodeRequest
-			// requires DecodeHeaders defined. Currently, we just documentate it. Per request check
-			// is expensive and not necessary in most of time.
+			filters[i] = model.NewFilterWrapper(fc.Name, f)
 		}
-		filters[i] = model.NewFilterWrapper(fc.Name, f)
-	}
 
-	if fm.canSkipMethod == nil {
-		fm.canSkipMethod = canSkipMethod
-	}
+		if fm.canSkipMethod == nil {
+			fm.canSkipMethod = canSkipMethod
+		}
 
-	// We can't cache the slice of filters as it may be changed by consumer
-	fm.filters = filters
+		// We can't cache the slice of filters as it may be changed by consumer
+		fm.filters = filters
 
-	if conf.consumerFiltersEndAt != 0 {
-		consumerFiltersEndAt := conf.consumerFiltersEndAt
-		consumerFilters := filters[:consumerFiltersEndAt]
-		fm.consumerFilters = consumerFilters
-		fm.filters = filters[consumerFiltersEndAt:]
-	}
+		if conf.consumerFiltersEndAt != 0 {
+			consumerFiltersEndAt := conf.consumerFiltersEndAt
+			consumerFilters := filters[:consumerFiltersEndAt]
+			fm.consumerFilters = consumerFilters
+			fm.filters = filters[consumerFiltersEndAt:]
+		}
 
-	// The skip check is based on the compiled code. So if the DecodeRequest is defined,
-	// even it is not called, DecodeData will not be skipped. Same as EncodeResponse.
-	fm.canSkipDecodeHeaders = fm.canSkipMethod["DecodeHeaders"]
-	fm.canSkipDecodeData = fm.canSkipMethod["DecodeData"] && fm.canSkipMethod["DecodeRequest"]
-	fm.canSkipEncodeHeaders = fm.canSkipMethod["EncodeHeaders"]
-	fm.canSkipEncodeData = fm.canSkipMethod["EncodeData"] && fm.canSkipMethod["EncodeResponse"]
-	fm.canSkipOnLog = fm.canSkipMethod["OnLog"]
+		// The skip check is based on the compiled code. So if the DecodeRequest is defined,
+		// even it is not called, DecodeData will not be skipped. Same as EncodeResponse.
+		fm.canSkipDecodeHeaders = fm.canSkipMethod["DecodeHeaders"]
+		fm.canSkipDecodeData = fm.canSkipMethod["DecodeData"] && fm.canSkipMethod["DecodeRequest"]
+		fm.canSkipEncodeHeaders = fm.canSkipMethod["EncodeHeaders"]
+		fm.canSkipEncodeData = fm.canSkipMethod["EncodeData"] && fm.canSkipMethod["EncodeResponse"]
+		fm.canSkipOnLog = fm.canSkipMethod["OnLog"]
 
-	return fm
+		return fm
+	}
 }
 
 func (m *filterManager) handleAction(res api.ResultAction, phase phase) (needReturn bool) {

pkg/filtermanager/filtermanager_benchmark_test.go

@@ -43,7 +43,7 @@ func BenchmarkFilterManagerAllPhase(b *testing.B) {
 	respBuf := envoy.NewBufferInstance([]byte{})
 
 	for n := 0; n < b.N; n++ {
-		m := FilterManagerFactory(config, cb)
+		m := FilterManagerFactory(config)(cb)
 		m.DecodeHeaders(reqHdr, false)
 		cb.WaitContinued()
 		m.DecodeData(reqBuf, true)
@@ -85,7 +85,7 @@ func BenchmarkFilterManagerRegular(b *testing.B) {
 	reqHdr := envoy.NewRequestHeaderMap(http.Header{})
 
 	for n := 0; n < b.N; n++ {
-		m := FilterManagerFactory(config, cb)
+		m := FilterManagerFactory(config)(cb)
 		m.DecodeHeaders(reqHdr, false)
 		cb.WaitContinued()
 		m.OnLog()
@@ -126,7 +126,7 @@ func BenchmarkFilterManagerConsumerWithFilter(b *testing.B) {
 	}
 
 	for n := 0; n < b.N; n++ {
-		m := FilterManagerFactory(config, cb)
+		m := FilterManagerFactory(config)(cb)
 		m.DecodeHeaders(reqHdrs[n%num], false)
 		cb.WaitContinued()
 		m.OnLog()

pkg/filtermanager/filtermanager_test.go

@@ -86,7 +86,7 @@ func TestPassThrough(t *testing.T) {
 		},
 	}
 	for i := 0; i < 2; i++ {
-		m := FilterManagerFactory(config, cb).(*filterManager)
+		m := FilterManagerFactory(config)(cb).(*filterManager)
 		hdr := envoy.NewRequestHeaderMap(http.Header{})
 		m.DecodeHeaders(hdr, false)
 		cb.WaitContinued()
@@ -154,7 +154,7 @@ func TestLocalReplyJSON_UseReqHeader(t *testing.T) {
 					Factory: PassThroughFactory,
 				},
 			}
-			m := FilterManagerFactory(config, cb).(*filterManager)
+			m := FilterManagerFactory(config)(cb).(*filterManager)
 			patches := gomonkey.ApplyMethodReturn(m.filters[0].Filter, "DecodeHeaders", &api.LocalResponse{
 				Code: 200,
 				Msg:  "msg",
@@ -227,7 +227,7 @@ func TestLocalReplyJSON_UseRespHeader(t *testing.T) {
 					Factory: PassThroughFactory,
 				},
 			}
-			m := FilterManagerFactory(config, cb).(*filterManager)
+			m := FilterManagerFactory(config)(cb).(*filterManager)
 			patches := gomonkey.ApplyMethodReturn(m.filters[0].Filter, "EncodeHeaders", &api.LocalResponse{
 				Code: 200,
 				Msg:  "msg",
@@ -263,7 +263,7 @@ func TestLocalReplyJSON_DoNotChangeMsgIfContentTypeIsGiven(t *testing.T) {
 			Factory: PassThroughFactory,
 		},
 	}
-	m := FilterManagerFactory(config, cb).(*filterManager)
+	m := FilterManagerFactory(config)(cb).(*filterManager)
 	patches := gomonkey.ApplyMethodReturn(m.filters[0].Filter, "DecodeHeaders", &api.LocalResponse{
 		Msg:    "msg",
 		Header: http.Header(map[string][]string{"Content-Type": {"text/plain"}}),
@@ -357,7 +357,7 @@ func TestSkipMethodWhenThereAreMultiFilters(t *testing.T) {
 	}
 
 	for i := 0; i < 2; i++ {
-		m := FilterManagerFactory(config, cb).(*filterManager)
+		m := FilterManagerFactory(config)(cb).(*filterManager)
 		assert.Equal(t, false, m.canSkipOnLog)
 		assert.Equal(t, false, m.canSkipDecodeHeaders)
 		assert.Equal(t, true, m.canSkipDecodeData)
@@ -407,7 +407,7 @@ func TestFiltersFromConsumer(t *testing.T) {
 		},
 	}
 	for i := 0; i < 20; i++ {
-		m := FilterManagerFactory(config, cb).(*filterManager)
+		m := FilterManagerFactory(config)(cb).(*filterManager)
 		assert.Equal(t, true, m.canSkipOnLog)
 		assert.Equal(t, 1, len(m.filters))
 		h := http.Header{}
@@ -477,7 +477,7 @@ func TestPluginState(t *testing.T) {
 			Factory: getPluginStateFilterFactory,
 		},
 	}
-	m := FilterManagerFactory(config, cb).(*filterManager)
+	m := FilterManagerFactory(config)(cb).(*filterManager)
 	h := http.Header{}
 	hdr := envoy.NewRequestHeaderMap(h)
 	m.DecodeHeaders(hdr, true)

plugins/tests/integration/data_plane/data_plane.go

@@ -138,10 +138,10 @@ func StartDataPlane(t *testing.T, opt *Option) (*DataPlane, error) {
 		// Since we only care about the coverage in CI, it is fine so far.
 	}
 
-	// This is the envoyproxy/envoy:contrib-dev fetched in 2024-02-08
-	// Use docker inspect --format='{{index .RepoDigests 0}}' envoyproxy/envoy:contrib-dev
+	// This is the envoyproxy/envoy:contrib-v1.29-latest
+	// Use docker inspect --format='{{index .RepoDigests 0}}' envoyproxy/envoy:contrib-v1.29-latest
 	// to get the sha256 ID
-	image := "envoyproxy/envoy@sha256:4ac92ea7095c787b6d1d74ea7d27a5c776ded08a0dee9b6737f2105f790fa384"
+	image := "envoyproxy/envoy@sha256:98ed3d86ff8b86dc12ddf54b7bb67ddf5506f80769038b3e2ab7bf402730fb4d"
 	pwd, _ := os.Getwd()
 	projectRoot := filepath.Join(pwd, "..", "..", "..")
 	cmdline := "docker run" +

plugins/tests/integration/libgolang/main.go

@@ -26,8 +26,8 @@ import (
 )
 
 func init() {
-	http.RegisterHttpFilterFactoryAndConfigParser("fm", filtermanager.FilterManagerFactory, &filtermanager.FilterManagerConfigParser{})
-	http.RegisterHttpFilterFactoryAndConfigParser("cm", consumer.ConsumerManagerFactory, &consumer.ConsumerManagerConfigParser{})
+	http.RegisterHttpFilterConfigFactoryAndParser("fm", filtermanager.FilterManagerFactory, &filtermanager.FilterManagerConfigParser{})
+	http.RegisterHttpFilterConfigFactoryAndParser("cm", consumer.ConsumerManagerFactory, &consumer.ConsumerManagerConfigParser{})
 }
 
 func main() {}