e2e: add receive test with capnp replication

Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com>

pkg/receive/capnp_server.go

@@ -103,35 +103,6 @@ func (c CapNProtoHandler) Write(ctx context.Context, call writecapnp.Writer_writ
 			result.SetError(writecapnp.WriteError_internal)
 		}
 	}
-	return nil
-}
-
-type BufferedListener struct {
-	ctx    context.Context
-	cancel context.CancelFunc
-
-	conns chan net.Conn
-}
-
-func (b BufferedListener) Accept() (net.Conn, error) {
-	select {
-	case <-b.ctx.Done():
-		return nil, b.ctx.Err()
-	case c := <-b.conns:
-		return c, nil
-	}
-}
 
-func (b BufferedListener) Close() error {
-	b.cancel()
 	return nil
 }
-
-func (b BufferedListener) Addr() net.Addr {
-	return addr{}
-}
-
-type addr struct{}
-
-func (addr) Network() string { return "bufconn" }
-func (addr) String() string  { return "bufconn" }

test/e2e/e2ethanos/services.go

@@ -539,6 +539,7 @@ type ReceiveBuilder struct {
 	f e2e.FutureRunnable
 
 	maxExemplars        int
+	capnp               bool
 	ingestion           bool
 	limit               int
 	tenantsLimits       receive.TenantsWriteLimitsConfig
@@ -555,7 +556,7 @@ type ReceiveBuilder struct {
 
 func NewReceiveBuilder(e e2e.Environment, name string) *ReceiveBuilder {
 	f := e.Runnable(fmt.Sprintf("receive-%v", name)).
-		WithPorts(map[string]int{"http": 8080, "grpc": 9091, "remote-write": 8081}).
+		WithPorts(map[string]int{"http": 8080, "grpc": 9091, "remote-write": 8081, "capnp": 19391}).
 		Future()
 	return &ReceiveBuilder{
 		Linkable:    f,
@@ -586,6 +587,11 @@ func (r *ReceiveBuilder) WithLabel(name, value string) *ReceiveBuilder {
 	return r
 }
 
+func (r *ReceiveBuilder) UseCapnpReplication() *ReceiveBuilder {
+	r.capnp = true
+	return r
+}
+
 func (r *ReceiveBuilder) WithRouting(replication int, hashringConfigs ...receive.HashringConfig) *ReceiveBuilder {
 	r.hashringConfigs = hashringConfigs
 	r.replication = replication
@@ -646,6 +652,10 @@ func (r *ReceiveBuilder) Init() *e2eobs.Observable {
 		args["--label"] = fmt.Sprintf("%s,%s", args["--label"], strings.Join(r.labels, ","))
 	}
 
+	if r.capnp {
+		args["--receive.capnproto-replication"] = ""
+	}
+
 	hashring := r.hashringConfigs
 	if len(hashring) > 0 && r.ingestion {
 		args["--receive.local-endpoint"] = r.InternalEndpoint("grpc")

test/e2e/receive_test.go

@@ -35,6 +35,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/promclient"
 	"github.com/thanos-io/thanos/pkg/receive"
 	"github.com/thanos-io/thanos/pkg/runutil"
+	"github.com/thanos-io/thanos/pkg/store/labelpb"
 	"github.com/thanos-io/thanos/test/e2e/e2ethanos"
 )
 
@@ -1152,3 +1153,45 @@ func TestReceiveExtractsTenant(t *testing.T) {
 
 	})
 }
+
+func TestReceiveCpnp(t *testing.T) {
+	e, err := e2e.NewDockerEnvironment("receive-cpnp")
+	testutil.Ok(t, err)
+	t.Cleanup(e2ethanos.CleanScenario(t, e))
+
+	i := e2ethanos.NewReceiveBuilder(e, "ingestor").WithIngestionEnabled().Init()
+	testutil.Ok(t, e2e.StartAndWaitReady(i))
+
+	h := receive.HashringConfig{
+		TenantMatcherType: "glob",
+		Tenants: []string{
+			"default*",
+		},
+		Endpoints: []receive.Endpoint{
+			{Address: i.InternalEndpoint("grpc"), CapNProtoAddress: i.InternalEndpoint("capnp")},
+		},
+	}
+
+	r := e2ethanos.NewReceiveBuilder(e, "router").WithRouting(1, h).UseCapnpReplication().Init()
+	testutil.Ok(t, e2e.StartAndWaitReady(r))
+
+	q := e2ethanos.NewQuerierBuilder(e, "1", i.InternalEndpoint("grpc")).Init()
+	testutil.Ok(t, e2e.StartAndWaitReady(q))
+
+	require.NoError(t, runutil.RetryWithLog(logkit.NewLogfmtLogger(os.Stdout), 1*time.Second, make(<-chan struct{}), func() error {
+		return storeWriteRequest(context.Background(), "http://"+r.Endpoint("remote-write")+"/api/v1/receive", &prompb.WriteRequest{
+			Timeseries: []*prompb.TimeSeries{
+				{
+					Labels: []*labelpb.Label{
+						{Name: "aa", Value: "bb"},
+					},
+					Samples: []*prompb.Sample{
+						{Value: 1, Timestamp: time.Now().UnixMilli()},
+					},
+				},
+			},
+		})
+	}))
+
+	testutil.Ok(t, i.WaitSumMetricsWithOptions(e2emon.Equals(0), []string{"prometheus_tsdb_blocks_loaded"}, e2emon.WithLabelMatchers(matchers.MustNewMatcher(matchers.MatchEqual, "tenant", "default-tenant")), e2emon.WaitMissingMetrics()))
+}