feat(packages): add registry and http packages (SigNoz#5740)

### Summary

Add packages for Registry and HTTP

#### Related Issues / PR's

SigNoz#5710

.gitignore

@@ -67,3 +67,6 @@ e2e/.auth
 # go
 vendor/
 **/main/**
+
+# git-town
+.git-branches.toml

ee/query-service/app/server.go

@@ -376,6 +376,7 @@ func (s *Server) createPublicServer(apiHandler *api.APIHandler) (*http.Server, e
 	}, nil
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/logging.go
 // loggingMiddleware is used for logging public api calls
 func loggingMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -387,6 +388,7 @@ func loggingMiddleware(next http.Handler) http.Handler {
 	})
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/logging.go
 // loggingMiddlewarePrivate is used for logging private api calls
 // from internal services like alert manager
 func loggingMiddlewarePrivate(next http.Handler) http.Handler {
@@ -399,27 +401,32 @@ func loggingMiddlewarePrivate(next http.Handler) http.Handler {
 	})
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/logging.go
 type loggingResponseWriter struct {
 	http.ResponseWriter
 	statusCode int
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/logging.go
 func NewLoggingResponseWriter(w http.ResponseWriter) *loggingResponseWriter {
 	// WriteHeader(int) is not called if our response implicitly returns 200 OK, so
 	// we default to that status code.
 	return &loggingResponseWriter{w, http.StatusOK}
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/logging.go
 func (lrw *loggingResponseWriter) WriteHeader(code int) {
 	lrw.statusCode = code
 	lrw.ResponseWriter.WriteHeader(code)
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/logging.go
 // Flush implements the http.Flush interface.
 func (lrw *loggingResponseWriter) Flush() {
 	lrw.ResponseWriter.(http.Flusher).Flush()
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/logging.go
 // Support websockets
 func (lrw *loggingResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
 	h, ok := lrw.ResponseWriter.(http.Hijacker)
@@ -565,6 +572,7 @@ func (s *Server) analyticsMiddleware(next http.Handler) http.Handler {
 	})
 }
 
+// TODO(remove): Implemented at pkg/http/middleware/timeout.go
 func setTimeoutMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()

pkg/http/doc.go

@@ -0,0 +1,3 @@
+// package http contains all http related functions such
+// as servers, middlewares, routers and renders.
+package http

pkg/http/middleware/doc.go

@@ -0,0 +1,2 @@
+// package middleware contains an implementation of all middlewares.
+package middleware

pkg/http/middleware/logging.go

@@ -0,0 +1,72 @@
+package middleware
+
+import (
+	"bytes"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/mux"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
+	"go.uber.org/zap"
+)
+
+const (
+	logMessage string = "::RECEIVED-REQUEST::"
+)
+
+type Logging struct {
+	logger *zap.Logger
+}
+
+func NewLogging(logger *zap.Logger) *Logging {
+	if logger == nil {
+		panic("cannot build logging, logger is empty")
+	}
+
+	return &Logging{
+		logger: logger.Named(pkgname),
+	}
+}
+
+func (middleware *Logging) Wrap(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		ctx := req.Context()
+		start := time.Now()
+		host, port, _ := net.SplitHostPort(req.Host)
+		path, err := mux.CurrentRoute(req).GetPathTemplate()
+		if err != nil {
+			path = req.URL.Path
+		}
+
+		fields := []zap.Field{
+			zap.Any("context", ctx),
+			zap.String(string(semconv.ClientAddressKey), req.RemoteAddr),
+			zap.String(string(semconv.UserAgentOriginalKey), req.UserAgent()),
+			zap.String(string(semconv.ServerAddressKey), host),
+			zap.String(string(semconv.ServerPortKey), port),
+			zap.Int64(string(semconv.HTTPRequestSizeKey), req.ContentLength),
+			zap.String(string(semconv.HTTPRouteKey), path),
+		}
+
+		buf := new(bytes.Buffer)
+		writer := newBadResponseLoggingWriter(rw, buf)
+		next.ServeHTTP(writer, req)
+
+		statusCode, err := writer.StatusCode(), writer.WriteError()
+		fields = append(fields,
+			zap.Int(string(semconv.HTTPResponseStatusCodeKey), statusCode),
+			zap.Duration(string(semconv.HTTPServerRequestDurationName), time.Since(start)),
+		)
+		if err != nil {
+			fields = append(fields, zap.Error(err))
+			middleware.logger.Error(logMessage, fields...)
+		} else {
+			if buf.Len() != 0 {
+				fields = append(fields, zap.String("response.body", buf.String()))
+			}
+
+			middleware.logger.Info(logMessage, fields...)
+		}
+	})
+}

pkg/http/middleware/middleware.go

@@ -0,0 +1,20 @@
+package middleware
+
+import "net/http"
+
+const (
+	pkgname string = "go.signoz.io/pkg/http/middleware"
+)
+
+// Wrapper is an interface implemented by all middlewares
+type Wrapper interface {
+	Wrap(http.Handler) http.Handler
+}
+
+// WrapperFunc is to Wrapper as http.HandlerFunc is to http.Handler
+type WrapperFunc func(http.Handler) http.Handler
+
+// WrapperFunc implements Wrapper
+func (m WrapperFunc) Wrap(next http.Handler) http.Handler {
+	return m(next)
+}

pkg/http/middleware/response.go

@@ -0,0 +1,122 @@
+package middleware
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+)
+
+const (
+	maxResponseBodyInLogs = 4096 // At most 4k bytes from response bodies in our logs.
+)
+
+type badResponseLoggingWriter interface {
+	http.ResponseWriter
+	// Get the status code.
+	StatusCode() int
+	// Get the error while writing.
+	WriteError() error
+}
+
+func newBadResponseLoggingWriter(rw http.ResponseWriter, buffer io.Writer) badResponseLoggingWriter {
+	b := nonFlushingBadResponseLoggingWriter{
+		rw:            rw,
+		buffer:        buffer,
+		logBody:       false,
+		bodyBytesLeft: maxResponseBodyInLogs,
+		statusCode:    http.StatusOK,
+	}
+
+	if f, ok := rw.(http.Flusher); ok {
+		return &flushingBadResponseLoggingWriter{b, f}
+	}
+
+	return &b
+}
+
+type nonFlushingBadResponseLoggingWriter struct {
+	rw            http.ResponseWriter
+	buffer        io.Writer
+	logBody       bool
+	bodyBytesLeft int
+	statusCode    int
+	writeError    error // The error returned when downstream Write() fails.
+}
+
+// Extends nonFlushingBadResponseLoggingWriter that implements http.Flusher
+type flushingBadResponseLoggingWriter struct {
+	nonFlushingBadResponseLoggingWriter
+	f http.Flusher
+}
+
+// Unwrap method is used by http.ResponseController to get access to original http.ResponseWriter.
+func (writer *nonFlushingBadResponseLoggingWriter) Unwrap() http.ResponseWriter {
+	return writer.rw
+}
+
+// Header returns the header map that will be sent by WriteHeader.
+// Implements ResponseWriter.
+func (writer *nonFlushingBadResponseLoggingWriter) Header() http.Header {
+	return writer.rw.Header()
+}
+
+// WriteHeader writes the HTTP response header.
+func (writer *nonFlushingBadResponseLoggingWriter) WriteHeader(statusCode int) {
+	writer.statusCode = statusCode
+	if statusCode >= 500 || statusCode == 400 {
+		writer.logBody = true
+	}
+	writer.rw.WriteHeader(statusCode)
+}
+
+// Writes HTTP response data.
+func (writer *nonFlushingBadResponseLoggingWriter) Write(data []byte) (int, error) {
+	if writer.statusCode == 0 {
+		// WriteHeader has (probably) not been called, so we need to call it with StatusOK to fulfill the interface contract.
+		// https://godoc.org/net/http#ResponseWriter
+		writer.WriteHeader(http.StatusOK)
+	}
+	n, err := writer.rw.Write(data)
+	if writer.logBody {
+		writer.captureResponseBody(data)
+	}
+	if err != nil {
+		writer.writeError = err
+	}
+	return n, err
+}
+
+// Hijack hijacks the first response writer that is a Hijacker.
+func (writer *nonFlushingBadResponseLoggingWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	hj, ok := writer.rw.(http.Hijacker)
+	if ok {
+		return hj.Hijack()
+	}
+	return nil, nil, fmt.Errorf("cannot cast underlying response writer to Hijacker")
+}
+
+func (writer *nonFlushingBadResponseLoggingWriter) StatusCode() int {
+	return writer.statusCode
+}
+
+func (writer *nonFlushingBadResponseLoggingWriter) WriteError() error {
+	return writer.writeError
+}
+
+func (writer *flushingBadResponseLoggingWriter) Flush() {
+	writer.f.Flush()
+}
+
+func (writer *nonFlushingBadResponseLoggingWriter) captureResponseBody(data []byte) {
+	if len(data) > writer.bodyBytesLeft {
+		_, _ = writer.buffer.Write(data[:writer.bodyBytesLeft])
+		_, _ = io.WriteString(writer.buffer, "...")
+		writer.bodyBytesLeft = 0
+		writer.logBody = false
+	} else {
+		_, _ = writer.buffer.Write(data)
+		writer.bodyBytesLeft -= len(data)
+	}
+}

pkg/http/middleware/timeout.go

@@ -0,0 +1,78 @@
+package middleware
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+const (
+	headerName string = "timeout"
+)
+
+type Timeout struct {
+	logger   *zap.Logger
+	excluded map[string]struct{}
+	// The default timeout
+	defaultTimeout time.Duration
+	// The max allowed timeout
+	maxTimeout time.Duration
+}
+
+func NewTimeout(logger *zap.Logger, excluded map[string]struct{}, defaultTimeout time.Duration, maxTimeout time.Duration) *Timeout {
+	if logger == nil {
+		panic("cannot build timeout, logger is empty")
+	}
+
+	if excluded == nil {
+		excluded = make(map[string]struct{})
+	}
+
+	if defaultTimeout.Seconds() == 0 {
+		defaultTimeout = 60 * time.Second
+	}
+
+	if maxTimeout == 0 {
+		maxTimeout = 600 * time.Second
+	}
+
+	return &Timeout{
+		logger:         logger.Named(pkgname),
+		excluded:       excluded,
+		defaultTimeout: defaultTimeout,
+		maxTimeout:     maxTimeout,
+	}
+}
+
+func (middleware *Timeout) Wrap(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		if _, ok := middleware.excluded[req.URL.Path]; !ok {
+			actual := middleware.defaultTimeout
+			incoming := req.Header.Get(headerName)
+			if incoming != "" {
+				parsed, err := time.ParseDuration(strings.TrimSpace(incoming) + "s")
+				if err != nil {
+					middleware.logger.Warn("cannot parse timeout in header, using default timeout", zap.String("timeout", incoming), zap.Error(err), zap.Any("context", req.Context()))
+				} else {
+					if parsed > middleware.maxTimeout {
+						actual = middleware.maxTimeout
+					} else {
+						actual = parsed
+					}
+				}
+			}
+
+			ctx, cancel := context.WithTimeout(req.Context(), actual)
+			defer cancel()
+
+			req = req.WithContext(ctx)
+			next.ServeHTTP(rw, req)
+			return
+		}
+
+		next.ServeHTTP(rw, req)
+	})
+}