increment 13

go.mod

@@ -3,15 +3,76 @@ module github.com/nekr0z/muhame
 go 1.22.11
 
 require (
+	github.com/caarlos0/env/v11 v11.3.1
 	github.com/go-chi/chi/v5 v5.2.0
+	github.com/go-resty/resty/v2 v2.16.5
+	github.com/golang-migrate/migrate/v4 v4.18.2
+	github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa
+	github.com/jackc/pgx/v5 v5.7.2
 	github.com/stretchr/testify v1.10.0
+	github.com/testcontainers/testcontainers-go v0.35.0
+	github.com/testcontainers/testcontainers-go/modules/postgres v0.35.0
 	go.uber.org/zap v1.27.0
 )
 
 require (
-	github.com/caarlos0/env/v11 v11.3.1 // indirect
+	dario.cat/mergo v1.0.0 // indirect
+	github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/containerd/platforms v0.2.1 // indirect
+	github.com/cpuguy83/dockercfg v0.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/distribution/reference v0.6.0 // indirect
+	github.com/docker/docker v27.2.0+incompatible // indirect
+	github.com/docker/go-connections v0.5.0 // indirect
+	github.com/docker/go-units v0.5.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	github.com/klauspost/compress v1.17.4 // indirect
+	github.com/lib/pq v1.10.9 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/patternmatcher v0.6.0 // indirect
+	github.com/moby/sys/sequential v0.5.0 // indirect
+	github.com/moby/sys/user v0.1.0 // indirect
+	github.com/moby/sys/userns v0.1.0 // indirect
+	github.com/moby/term v0.5.0 // indirect
+	github.com/morikuni/aec v1.0.0 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.0 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
+	github.com/shirou/gopsutil/v3 v3.23.12 // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.3 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.54.0 // indirect
+	go.opentelemetry.io/otel v1.29.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 // indirect
+	go.opentelemetry.io/otel/metric v1.29.0 // indirect
+	go.opentelemetry.io/otel/trace v1.29.0 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.10.0 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

internal/agent/queue.go

@@ -1,8 +1,10 @@
 package agent
 
 import (
+	"bufio"
 	"bytes"
 	"compress/gzip"
+	"fmt"
 	"io"
 	"net/http"
 	"strings"
@@ -50,17 +52,108 @@ func (q *queue) pop() *queuedMetric {
 }
 
 func (q *queue) sendMetrics(c *http.Client, addr string) {
+	mm := make([]queuedMetric, 0)
+
 	for m := q.pop(); m != nil; m = q.pop() {
-		sendMetric(c, *m, addr)
+		mm = append(mm, *m)
+	}
+
+	if len(mm) == 0 {
+		return
+	}
+
+	sendAll(c, addr, mm)
+}
+
+func sendAll(c *http.Client, addr string, mm []queuedMetric) {
+	if sendBulk(c, addr, mm) == nil {
+		return
+	}
+
+	for _, m := range mm {
+		sendMetric(c, m, addr)
+	}
+}
+
+func sendBulk(c *http.Client, addr string, mm []queuedMetric) error {
+	b := zipBulk(mm)
+
+	req, err := http.NewRequest(http.MethodPost, endpointBulk(addr), b)
+	if err != nil {
+		panic(err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Content-Encoding", "gzip")
+
+	resp, err := c.Do(req)
+	if err != nil {
+		return err
+	}
+
+	if resp == nil {
+		return fmt.Errorf("nil response")
+	}
+
+	_, _ = io.Copy(io.Discard, resp.Body)
+	resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("bulk not accepted")
+	}
+
+	return nil
+}
+
+func zipBulk(mm []queuedMetric) *bytes.Buffer {
+	var b bytes.Buffer
+
+	w, err := gzip.NewWriterLevel(&b, gzip.BestCompression)
+	if err != nil {
+		panic(err)
+	}
+	defer w.Close()
+
+	bw := bufio.NewWriter(w)
+
+	_, err = bw.WriteRune('[')
+	if err != nil {
+		panic(err)
+	}
+
+	for i, m := range mm {
+		if i != 0 {
+			_, err = bw.WriteRune(',')
+			if err != nil {
+				panic(err)
+			}
+		}
+
+		_, err = bw.Write(metrics.ToJSON(m.val, m.name))
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	_, err = bw.WriteRune(']')
+	if err != nil {
+		panic(err)
 	}
+
+	err = bw.Flush()
+	if err != nil {
+		panic(err)
+	}
+
+	return &b
 }
 
 func sendMetric(c *http.Client, m queuedMetric, addr string) {
 	bb := metrics.ToJSON(m.val, m.name)
 
 	b := compress(bb)
 
-	req, err := http.NewRequest(http.MethodPost, endpoint(addr), &b)
+	req, err := http.NewRequest(http.MethodPost, endpointSingle(addr), &b)
 	if err != nil {
 		panic(err)
 	}
@@ -86,10 +179,14 @@ type queuedMetric struct {
 	next *queuedMetric
 }
 
-func endpoint(addr string) string {
+func endpointSingle(addr string) string {
 	return strings.TrimSuffix(addr, "/") + "/update/"
 }
 
+func endpointBulk(addr string) string {
+	return strings.TrimSuffix(addr, "/") + "/updates/"
+}
+
 func compress(b []byte) bytes.Buffer {
 	var buf bytes.Buffer
 

internal/agent/queue_test.go

@@ -62,3 +62,92 @@ func TestSendMetric(t *testing.T) {
 		})
 	}
 }
+
+func TestSendBulk(t *testing.T) {
+	mm := []queuedMetric{
+		{
+			name: "test",
+			val:  metrics.Gauge(1.2),
+		},
+		{
+			name: "another",
+			val:  metrics.Counter(2),
+		},
+	}
+
+	want := `[
+{"id": "test", "type": "gauge", "value": 1.2},
+{"id": "another", "type": "counter", "delta": 2}
+]`
+
+	q := queue{}
+
+	for _, m := range mm {
+		q.push(m)
+	}
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "/updates/", r.URL.Path)
+		assert.Equal(t, "application/json", r.Header.Get("Content-Type"))
+		assert.Equal(t, "gzip", r.Header.Get("Content-Encoding"))
+
+		b := r.Body
+		defer b.Close()
+
+		var err error
+		b, err = gzip.NewReader(b)
+		assert.NoError(t, err)
+
+		bb, err := io.ReadAll(b)
+		assert.NoError(t, err)
+
+		assert.JSONEq(t, want, string(bb))
+	}))
+	defer srv.Close()
+
+	q.sendMetrics(http.DefaultClient, srv.URL)
+}
+
+func TestSendBulk_Fallback(t *testing.T) {
+	mm := []queuedMetric{
+		{
+			name: "test",
+			val:  metrics.Gauge(1.2),
+		},
+		{
+			name: "another",
+			val:  metrics.Counter(2),
+		},
+	}
+
+	q := queue{}
+
+	for _, m := range mm {
+		q.push(m)
+	}
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/updates/" {
+			http.Error(w, "storage does not support bulk updates", http.StatusConflict)
+			return
+		}
+
+		assert.Equal(t, "/update/", r.URL.Path)
+		assert.Equal(t, "application/json", r.Header.Get("Content-Type"))
+		assert.Equal(t, "gzip", r.Header.Get("Content-Encoding"))
+	}))
+	defer srv.Close()
+
+	q.sendMetrics(http.DefaultClient, srv.URL)
+}
+
+func TestSendBulk_EmptyQueue(t *testing.T) {
+	q := queue{}
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		t.Fatal("should not send metrics from empty queue")
+	}))
+	defer srv.Close()
+
+	q.sendMetrics(http.DefaultClient, srv.URL)
+}

internal/handlers/ping.go

@@ -0,0 +1,31 @@
+package handlers
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/nekr0z/muhame/internal/storage"
+)
+
+func PingHandleFunc(st storage.Storage) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		p, ok := st.(pingable)
+		if !ok {
+			http.Error(w, "pinging the used storage makes no sense", http.StatusConflict)
+			return
+		}
+
+		err := p.Ping(r.Context())
+
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		w.WriteHeader(http.StatusOK)
+	}
+}
+
+type pingable interface {
+	Ping(context.Context) error
+}

internal/handlers/root.go

@@ -2,6 +2,7 @@ package handlers
 
 import (
 	"cmp"
+	"context"
 	"fmt"
 	"net/http"
 	"slices"
@@ -12,7 +13,7 @@ import (
 // RootHandleFunc returns the handler for the / endpoint.
 func RootHandleFunc(st storage.Storage) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
-		mm, err := listAllMetrics(st)
+		mm, err := listAllMetrics(r.Context(), st)
 		if err != nil {
 			http.Error(w, fmt.Sprintf("Internal server error: %s", err), http.StatusInternalServerError)
 			return
@@ -50,18 +51,18 @@ type displayedMetric struct {
 	value string
 }
 
-func listAllMetrics(st storage.Storage) ([]displayedMetric, error) {
-	names, mm, err := st.List()
+func listAllMetrics(ctx context.Context, st storage.Storage) ([]displayedMetric, error) {
+	mm, err := st.List(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	mms := make([]displayedMetric, 0, len(names))
-	for i, name := range names {
+	mms := make([]displayedMetric, 0, len(mm))
+	for _, named := range mm {
 		mms = append(mms, displayedMetric{
-			name:  name,
-			t:     mm[i].Type(),
-			value: mm[i].String(),
+			name:  named.Name,
+			t:     named.Type(),
+			value: named.String(),
 		})
 	}