[Feature] Add proper Prometheus endpoint compression (#1393)

Author: ajanikow

CHANGELOG.md

@@ -10,7 +10,8 @@
 - (Bugfix) Fix Member Terminating state discovery
 - (Bugfix) Fix CRD yaml (chart)
 - (Bugfix) (EE) Fix MemberMaintenance Context and ClusterMaintenance discovery
-- 
+- (Feature) Add proper Prometheus endpoint compression + 204 response code
+
 ## [1.2.32](https://github.com/arangodb/kube-arangodb/tree/1.2.32) (2023-08-07)
 - (Feature) Backup lifetime - remove Backup once its lifetime has been reached
 - (Feature) Add Feature dependency

pkg/deployment/features/local.go

@@ -30,7 +30,6 @@ import (
 
 	"github.com/arangodb/go-driver"
 
-	"github.com/arangodb/kube-arangodb/pkg/logging"
 	"github.com/arangodb/kube-arangodb/pkg/util"
 )
 
@@ -120,54 +119,9 @@ func Init(cmd *cobra.Command) error {
 
 	f.StringVar(&configMapName, "features-config-map-name", DefaultFeaturesConfigMap, "Name of the Feature Map ConfigMap")
 
-	checkDependencies(cmd)
-
 	return nil
 }
 
-func checkDependencies(cmd *cobra.Command) {
-
-	enableDeps := func(_ *cobra.Command, _ []string) {
-		// Turn on dependencies. This function will be called when all process's arguments are passed, so
-		// all required features are enabled and dependencies should be enabled too.
-		EnableDependencies()
-
-		// Log enabled features when process starts.
-		for _, f := range features {
-			if !f.Enabled() {
-				continue
-			}
-
-			l := logging.Global().RegisterAndGetLogger("features", logging.Info)
-			if deps := f.GetDependencies(); len(deps) > 0 {
-				l = l.Strs("dependencies", deps...)
-			}
-
-			l.Bool("enterpriseArangoDBRequired", f.EnterpriseRequired()).
-				Str("minArangoDBVersion", string(f.Version())).
-				Str("name", f.Name()).
-				Info("feature enabled")
-		}
-	}
-
-	// Wrap pre-run function if it set.
-	if cmd.PreRunE != nil {
-		local := cmd.PreRunE
-		cmd.PreRunE = func(cmd *cobra.Command, args []string) error {
-			enableDeps(cmd, args)
-			return local(cmd, args)
-		}
-	} else if cmd.PreRun != nil {
-		local := cmd.PreRun
-		cmd.PreRun = func(cmd *cobra.Command, args []string) {
-			enableDeps(cmd, args)
-			local(cmd, args)
-		}
-	} else {
-		cmd.PreRun = enableDeps
-	}
-}
-
 func cmdRun(_ *cobra.Command, _ []string) {
 	featuresLock.Lock()
 	defer featuresLock.Unlock()

pkg/server/server.go

@@ -29,7 +29,6 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/jessevdk/go-assets"
-	prometheus "github.com/prometheus/client_golang/prometheus/promhttp"
 	core "k8s.io/api/core/v1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	typedCore "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -39,6 +38,7 @@ import (
 	"github.com/arangodb/kube-arangodb/dashboard"
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 	operatorHTTP "github.com/arangodb/kube-arangodb/pkg/util/http"
+	"github.com/arangodb/kube-arangodb/pkg/util/metrics"
 	"github.com/arangodb/kube-arangodb/pkg/util/probe"
 	"github.com/arangodb/kube-arangodb/pkg/version"
 )
@@ -177,7 +177,7 @@ func NewServer(cli typedCore.CoreV1Interface, cfg Config, deps Dependencies) (*S
 		readyProbes = append(readyProbes, deps.Storage.Probe)
 	}
 	r.GET("/ready", gin.WrapF(ready(readyProbes...)))
-	r.GET("/metrics", gin.WrapH(prometheus.Handler()))
+	r.GET("/metrics", gin.WrapF(metrics.Handler()))
 	r.POST("/login", s.auth.handleLogin)
 	api := r.Group("/api", s.auth.checkAuthentication)
 	{

pkg/util/http/buffer.go

@@ -0,0 +1,163 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package http
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+
+	"github.com/arangodb/kube-arangodb/pkg/util"
+)
+
+const (
+	ContentLengthHeader = "Content-Length"
+)
+
+func WithBuffer(maxSize int, in http.HandlerFunc) http.HandlerFunc {
+	return func(writer http.ResponseWriter, request *http.Request) {
+		data := NewBuffer(maxSize, writer)
+
+		wr := NewWriter(writer, data)
+
+		in(wr, request)
+
+		if !data.Truncated() {
+			// We have constant size
+			bytes := data.Bytes()
+
+			println(len(bytes))
+
+			writer.Header().Set(ContentLengthHeader, fmt.Sprintf("%d", len(bytes)))
+
+			_, err := util.WriteAll(writer, bytes)
+			if err != nil {
+				logger.Err(err).Warn("Unable to write HTTP response")
+			}
+		}
+	}
+}
+
+type Buffer interface {
+	io.Writer
+
+	Bytes() []byte
+	Truncated() bool
+}
+
+type buffer struct {
+	lock sync.Mutex
+
+	upstream io.Writer
+
+	data, currentData []byte
+}
+
+func (b *buffer) Write(q []byte) (n int, err error) {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	p := q
+
+	for {
+		if len(p) == 0 || len(b.currentData) == 0 {
+			break
+		}
+
+		b.currentData[0] = p[0]
+		b.currentData = b.currentData[1:]
+		p = p[1:]
+	}
+
+	if len(p) == 0 {
+		return len(q), nil
+	}
+
+	written := 0
+
+	if len(b.currentData) == 0 {
+		if b.data != nil {
+			z, err := util.WriteAll(b.upstream, b.data)
+			if err != nil {
+				return 0, err
+			}
+
+			written += z
+
+			b.data = nil
+			b.currentData = nil
+		}
+	} else {
+		return len(q), nil
+	}
+
+	z, err := b.upstream.Write(p)
+	if err != nil {
+		return 0, err
+	}
+
+	written += z
+
+	return written, nil
+}
+
+func (b *buffer) Bytes() []byte {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	if len(b.data) == 0 {
+		return nil
+	}
+
+	return b.data[:len(b.data)-len(b.currentData)]
+}
+
+func (b *buffer) Truncated() bool {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	return len(b.data) == 0
+}
+
+type bytesBuffer struct {
+	*bytes.Buffer
+}
+
+func (b bytesBuffer) Truncated() bool {
+	return false
+}
+
+func NewBuffer(maxSize int, upstream io.Writer) Buffer {
+	if maxSize <= 0 {
+		return &bytesBuffer{
+			Buffer: bytes.NewBuffer(nil),
+		}
+	}
+
+	b := &buffer{
+		data:     make([]byte, maxSize),
+		upstream: upstream,
+	}
+	b.currentData = b.data
+	return b
+}

pkg/util/http/buffer_test.go

@@ -0,0 +1,93 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package http
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/arangodb/kube-arangodb/pkg/util"
+)
+
+func Test_Buffer(t *testing.T) {
+	t.Run("Normal cache", func(t *testing.T) {
+		up := bytes.NewBuffer(nil)
+		b := NewBuffer(4096, up)
+
+		data := make([]byte, 1024)
+
+		c, err := util.WriteAll(b, data)
+		require.Len(t, data, c)
+
+		require.NoError(t, err)
+
+		require.Len(t, up.Bytes(), 0)
+		require.Len(t, b.Bytes(), 1024)
+		require.False(t, b.Truncated())
+	})
+	t.Run("Full cache", func(t *testing.T) {
+		up := bytes.NewBuffer(nil)
+		b := NewBuffer(1024, up)
+
+		data := make([]byte, 1024)
+
+		c, err := util.WriteAll(b, data)
+		require.Len(t, data, c)
+
+		require.NoError(t, err)
+
+		require.Len(t, up.Bytes(), 0)
+		require.Len(t, b.Bytes(), 1024)
+		require.False(t, b.Truncated())
+	})
+	t.Run("Full cache + 1", func(t *testing.T) {
+		up := bytes.NewBuffer(nil)
+		b := NewBuffer(1024, up)
+
+		data := make([]byte, 1025)
+
+		c, err := util.WriteAll(b, data)
+		require.Len(t, data, c)
+
+		require.NoError(t, err)
+
+		require.Len(t, up.Bytes(), 1025)
+		require.Len(t, b.Bytes(), 0)
+		require.True(t, b.Truncated())
+	})
+	t.Run("Overflow cache", func(t *testing.T) {
+		up := bytes.NewBuffer(nil)
+		b := NewBuffer(1024, up)
+
+		data := make([]byte, 2048)
+
+		c, err := util.WriteAll(b, data)
+		require.Len(t, data, c)
+
+		require.NoError(t, err)
+
+		require.Len(t, up.Bytes(), 2048)
+		require.Len(t, b.Bytes(), 0)
+		require.True(t, b.Truncated())
+	})
+}

pkg/util/http/content.go

@@ -0,0 +1,37 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package http
+
+import "net/http"
+
+const ContentTypeHeader = "Content-Type"
+
+func WithTextContentType(in http.HandlerFunc) http.HandlerFunc {
+	return WithContentType("plain/text", in)
+}
+
+func WithContentType(content string, in http.HandlerFunc) http.HandlerFunc {
+	return func(writer http.ResponseWriter, request *http.Request) {
+		writer.Header().Set(ContentTypeHeader, content)
+
+		in(writer, request)
+	}
+}