Feature/compression (#44)

* compression

go.mod

@@ -4,7 +4,7 @@ go 1.12
 
 require (
 	github.com/aws/aws-sdk-go v1.32.10
-	github.com/envoyproxy/go-control-plane v0.9.5
+	github.com/envoyproxy/go-control-plane v0.9.6-0.20200618221453-226baa5cddab
 	github.com/ghodss/yaml v1.0.0
 	github.com/gogo/protobuf v1.3.1
 	github.com/golang/protobuf v1.4.2

go.sum

@@ -25,6 +25,8 @@ github.com/envoyproxy/go-control-plane v0.9.2/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/go-control-plane v0.9.5 h1:lRJIqDD8yjV1YyPRqecMdytjDLs2fTXq363aCib5xPU=
 github.com/envoyproxy/go-control-plane v0.9.5/go.mod h1:OXl5to++W0ctG+EHWTFUjiypVxC/Y4VLc/KFU+al13s=
+github.com/envoyproxy/go-control-plane v0.9.6-0.20200618221453-226baa5cddab h1:ALuQRm7L5dY+9HEFlo5il+sENgolnfjWRHKSf8TgYf4=
+github.com/envoyproxy/go-control-plane v0.9.6-0.20200618221453-226baa5cddab/go.mod h1:JvuSsUgXzeWfLVfAe9OeW40eBtd+E8yMydqNm0iuBxs=
 github.com/envoyproxy/protoc-gen-validate v0.0.0-20190405222122-d6164de49109 h1:FNgqGzbOm637YKRbYGKb9cqGo8i50++w/LWvMau7jrw=
 github.com/envoyproxy/protoc-gen-validate v0.0.0-20190405222122-d6164de49109/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/envoyproxy/protoc-gen-validate v0.0.14 h1:YBW6/cKy9prEGRYLnaGa4IDhzxZhRCtKsax8srGKDnM=

pkg/api/compression.go

@@ -0,0 +1,14 @@
+package api
+
+type Compression struct {
+	API      string          `json:"api" yaml:"api"`
+	Kind     string          `json:"kind" yaml:"kind"`
+	Metadata Metadata        `json:"metadata" yaml:"metadata"`
+	Spec     CompressionSpec `json:"spec" yaml:"spec"`
+}
+type CompressionSpec struct {
+	Type                string   `json:"type" yaml:"type"`
+	ContentLength       uint32   `json:"contentLength" yaml:"contentLength"`
+	ContentType         []string `json:"contentType" yaml:"contentType"`
+	DisableOnEtagHeader bool     `json:"disableOnEtagHeader" yaml:"disableOnEtagHeader"`
+}

pkg/envoy/compression.go

@@ -0,0 +1,101 @@
+package envoy
+
+import (
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	api "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	gzip "github.com/envoyproxy/go-control-plane/envoy/extensions/compression/gzip/compressor/v3"
+	compressor "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/http/compressor/v3"
+	"github.com/golang/protobuf/ptypes"
+	any "github.com/golang/protobuf/ptypes/any"
+	"github.com/golang/protobuf/ptypes/wrappers"
+)
+
+type Compression struct{}
+
+func newCompression() *Compression {
+	return &Compression{}
+}
+
+func (c *Compression) updateListenersWithCompression(cache *WorkQueueCache, params CompressionParams) error {
+	// update listener
+	for listenerKey := range cache.listeners {
+		ll := cache.listeners[listenerKey].(*api.Listener)
+		for filterchainID := range ll.FilterChains {
+			for filterID := range ll.FilterChains[filterchainID].Filters {
+				// get manager
+				manager, err := getManager((ll.FilterChains[filterchainID].Filters[filterID].ConfigType).(*api.Filter_TypedConfig))
+				if err != nil {
+					return err
+				}
+
+				// get authz config config
+				compressorConfigEncoded, err := c.getCompressionFilterEncoded(params)
+				if err != nil {
+					return err
+				}
+
+				// update http filter
+				updateHTTPFilterWithConfig(&manager.HttpFilters, "envoy.filters.http.compressor", compressorConfigEncoded)
+
+				// update manager in cache
+				pbst, err := ptypes.MarshalAny(&manager)
+				if err != nil {
+					return err
+				}
+				ll.FilterChains[filterchainID].Filters[filterID].ConfigType = &api.Filter_TypedConfig{
+					TypedConfig: pbst,
+				}
+
+			}
+
+		}
+
+	}
+
+	return nil
+}
+
+func (c *Compression) getCompressionFilterEncoded(params CompressionParams) (*any.Any, error) {
+	compressionFilter, err := c.getCompressionFilter(params)
+	if compressionFilter == nil {
+		return nil, nil
+	}
+	compressionFilterEncoded, err := ptypes.MarshalAny(compressionFilter)
+	if err != nil {
+		return nil, err
+	}
+	return compressionFilterEncoded, nil
+}
+
+func (c *Compression) getCompressionFilter(compression CompressionParams) (*compressor.Compressor, error) {
+	if compression.Type == "gzip" {
+		// set gzip config
+		gzip := gzip.Gzip{
+			CompressionLevel:    gzip.Gzip_DEFAULT_COMPRESSION,
+			CompressionStrategy: gzip.Gzip_DEFAULT_STRATEGY,
+		}
+		gzipEncoded, err := ptypes.MarshalAny(&gzip)
+		if err != nil {
+			return nil, err
+		}
+		// set compressor config
+		httpFilterConfig := compressor.Compressor{
+			CompressorLibrary: &core.TypedExtensionConfig{
+				Name:        "text_optimized",
+				TypedConfig: gzipEncoded,
+			},
+		}
+		if compression.ContentLength != 0 {
+			httpFilterConfig.ContentLength = &wrappers.UInt32Value{
+				Value: compression.ContentLength,
+			}
+		}
+		if len(compression.ContentType) != 0 {
+			httpFilterConfig.ContentType = compression.ContentType
+		}
+		httpFilterConfig.DisableOnEtagHeader = compression.DisableOnEtagHeader
+
+		return &httpFilterConfig, nil
+	}
+	return nil, nil
+}

pkg/envoy/listener.go

@@ -700,6 +700,20 @@ func (l *Listener) updateDefaultTracingSetting(tracing TracingParams) {
 	}
 }
 
+func (l *Listener) updateDefaultCompressionSetting(compressionParams CompressionParams) {
+	c := newCompression()
+	compressorFilterEncoded, err := c.getCompressionFilterEncoded(compressionParams)
+	if err != nil {
+		logger.Errorf("Couldn't update default compression filter: %s", err)
+		return
+	}
+	if compressorFilterEncoded == nil {
+		return
+	}
+
+	updateHTTPFilterWithConfig(&l.httpFilter, "envoy.filters.http.compressor", compressorFilterEncoded)
+}
+
 func (l *Listener) newHTTPRouterFilter() []*hcm.HttpFilter {
 	return l.httpFilter
 }

pkg/envoy/testdata/test-compression.yaml

@@ -0,0 +1,7 @@
+api: proxy.in4it.io/v1
+kind: compression
+metadata:
+  name: compression
+spec:
+  type: gzip
+  disableOnEtagHeader: true

pkg/envoy/types.go

@@ -6,17 +6,18 @@ import (
 )
 
 type WorkQueueItem struct {
-	id               string
-	Action           string
-	DependsOn        string
-	DependsOnItemIDs []string
-	TLSParams        TLSParams
-	ClusterParams    ClusterParams
-	ListenerParams   ListenerParams
-	ChallengeParams  ChallengeParams
-	CreateCertParams CreateCertParams
-	TracingParams    TracingParams
-	state            string
+	id                string
+	Action            string
+	DependsOn         string
+	DependsOnItemIDs  []string
+	TLSParams         TLSParams
+	ClusterParams     ClusterParams
+	ListenerParams    ListenerParams
+	ChallengeParams   ChallengeParams
+	CreateCertParams  CreateCertParams
+	TracingParams     TracingParams
+	CompressionParams CompressionParams
+	state             string
 }
 
 type WorkQueueCache struct {
@@ -124,6 +125,13 @@ type TracingParams struct {
 	OverallSampling float64
 }
 
+type CompressionParams struct {
+	Type                string
+	ContentLength       uint32
+	ContentType         []string
+	DisableOnEtagHeader bool
+}
+
 type DirectResponse struct {
 	Status uint32
 	Body   string

pkg/envoy/workqueue.go

@@ -22,6 +22,7 @@ type WorkQueue struct {
 	jwtProvider    *JwtProvider
 	authzFilter    *AuthzFilter
 	tracing        *Tracing
+	compression    *Compression
 	cluster        *Cluster
 	acmeContact    string
 	latestSnapshot cache.Snapshot
@@ -49,6 +50,7 @@ func NewWorkQueue(s storage.Storage, acmeContact string) (*WorkQueue, error) {
 		jwtProvider: newJwtProvider(),
 		authzFilter: newAuthzFilter(),
 		tracing:     newTracing(),
+		compression: newCompression(),
 	}
 
 	// run queue to resolve dependencies
@@ -197,6 +199,18 @@ func (w *WorkQueue) Submit(items []WorkQueueItem) (string, error) {
 				item.state = "finished"
 			}
 			updateXds = true
+		case "updateListenersWithCompression":
+			// update default listener route
+			w.listener.updateDefaultCompressionSetting(item.CompressionParams)
+			// update existing listeners
+			err := w.compression.updateListenersWithCompression(&w.cache, item.CompressionParams)
+			if err != nil {
+				item.state = "error"
+				logger.Errorf("updateListenersWithCompression error: %s", err)
+			} else {
+				item.state = "finished"
+			}
+			updateXds = true
 		case "updateListenerWithChallenge":
 			err := w.listener.updateListenerWithChallenge(&w.cache, item.ChallengeParams)
 			if err != nil {

pkg/envoy/xds.go

@@ -219,6 +219,13 @@ func (x *XDS) ImportObject(object pkgApi.Object) ([]WorkQueueItem, error) {
 			return []WorkQueueItem{}, fmt.Errorf("Couldn't import new rule: %s", err)
 		}
 		return items, nil
+	case "compression":
+		compression := object.Data.(pkgApi.Compression)
+		items, err := x.importCompression(compression)
+		if err != nil {
+			return []WorkQueueItem{}, fmt.Errorf("Couldn't import new rule: %s", err)
+		}
+		return items, nil
 	}
 
 	return []WorkQueueItem{}, nil
@@ -262,6 +269,20 @@ func (x *XDS) importTracing(tracing pkgApi.Tracing) ([]WorkQueueItem, error) {
 	}, nil
 }
 
+func (x *XDS) importCompression(compression pkgApi.Compression) ([]WorkQueueItem, error) {
+	return []WorkQueueItem{
+		{
+			Action: "updateListenersWithCompression",
+			CompressionParams: CompressionParams{
+				Type:                compression.Spec.Type,
+				ContentLength:       compression.Spec.ContentLength,
+				ContentType:         compression.Spec.ContentType,
+				DisableOnEtagHeader: compression.Spec.DisableOnEtagHeader,
+			},
+		},
+	}, nil
+}
+
 func (x *XDS) importJwtProvider(jwtProvider pkgApi.JwtProvider) ([]WorkQueueItem, error) {
 	logger.Debugf("Found jwtProvider with name %s and jwksUrl %s", jwtProvider.Metadata.Name, jwtProvider.Spec.RemoteJwks)
 	u, err := url.Parse(jwtProvider.Spec.RemoteJwks)

pkg/envoy/xds_test.go

@@ -530,6 +530,7 @@ func TestClusterWithHealthcheck(t *testing.T) {
 	}
 	fmt.Printf("%s\n", out)
 }
+
 func TestClusterWithWebsockets(t *testing.T) {
 	logger.SetLogLevel(loggo.DEBUG)
 	s, err := initStorage()
@@ -587,3 +588,34 @@ func TestClusterWithWebsockets(t *testing.T) {
 		}
 	}
 }
+func TestCompressionObject(t *testing.T) {
+	logger.SetLogLevel(loggo.DEBUG)
+	s, err := initStorage()
+	if err != nil {
+		t.Errorf("Couldn't initialize storage: %s", err)
+		return
+	}
+	x := NewXDS(s, "", "")
+	ObjectFileNames := []string{"test-compression.yaml"}
+	for _, filename := range ObjectFileNames {
+		newItems, err := x.putObject(filename)
+		if err != nil {
+			t.Errorf("PutObject failed: %s", err)
+			return
+		}
+		_, err = x.workQueue.Submit(newItems)
+		if err != nil {
+			t.Errorf("WorkQueue error: %s", err)
+			return
+		}
+	}
+	httpFilters := x.workQueue.listener.newHTTPRouterFilter()
+	if len(httpFilters) == 0 {
+		t.Errorf("Filters in empty")
+		return
+	}
+	if httpFilters[0].Name != "envoy.filters.http.compressor" {
+		t.Errorf("Compressor filter not found")
+		return
+	}
+}

pkg/storage/local/io.go

@@ -140,6 +140,13 @@ func (l *LocalStorage) GetObject(name string) ([]api.Object, error) {
 					return objects, err
 				}
 				object.Data = tracing
+			case "compression":
+				var compression api.Compression
+				err = yaml.Unmarshal([]byte(contentsSplitted), &compression)
+				if err != nil {
+					return objects, err
+				}
+				object.Data = compression
 			default:
 				return objects, errors.New("Rule in wrong format")
 			}

pkg/storage/s3/io.go

@@ -169,6 +169,13 @@ func (s *S3Storage) GetObject(filename string) ([]api.Object, error) {
 					return objects, err
 				}
 				object.Data = tracing
+			case "compression":
+				var compression api.Compression
+				err = yaml.Unmarshal([]byte(contentsSplitted), &compression)
+				if err != nil {
+					return objects, err
+				}
+				object.Data = compression
 			default:
 				return objects, errors.New("Object in wrong format")
 			}

resources/envoy.yaml

@@ -6,19 +6,11 @@ dynamic_resources:
       envoy_grpc:
         cluster_name: xds_cluster
   cds_config:
+    ads: {}
     resource_api_version: V3
-    api_config_source:
-      api_type: GRPC
-      grpc_services:
-        envoy_grpc:
-          cluster_name: xds_cluster
   lds_config:
+    ads: {}
     resource_api_version: V3
-    api_config_source:
-      api_type: GRPC
-      grpc_services:
-        envoy_grpc:
-          cluster_name: xds_cluster
 node:
   cluster: ingress-gateway
   id: ingress-gateway-2