sidecar: Added support for streaming, chunked remote read.

Fixes: #488

Signed-off-by: Bartek Plotka <[email protected]>

.circleci/config.yml

@@ -23,6 +23,9 @@ jobs:
               echo "Awesome! GCS integration tests are enabled."
             fi
       - run: make deps
+      # Debugging lint.
+      - run: make golangclint
+      - run: /go/bin/golangci-lint-97ea1cbb21bbf5e4d0e8bcc0f9243385e9262dcc run -v ./...
       - run: make lint
       - run: make check-docs
       - run: make format
@@ -31,19 +34,10 @@ jobs:
           # TODO(bplotka): Setup some S3 tests for CI.
           command: |
             if [ -z ${GCP_PROJECT} ]; then
-              export THANOS_SKIP_GCS_TESTS="true"
-              echo "Skipping GCS tests."
+              make test-local
+              exit
             fi
-            export THANOS_SKIP_S3_AWS_TESTS="true"
-            echo "Skipping AWS tests."
-            export THANOS_SKIP_AZURE_TESTS="true"
-            echo "Skipping Azure tests."
-            export THANOS_SKIP_SWIFT_TESTS="true"
-            echo "Skipping SWIFT tests."
-            export THANOS_SKIP_TENCENT_COS_TESTS="true"
-            echo "Skipping TENCENT COS tests."
-
-            make test
+            make test-only-gcs
 
   # Cross build is needed for publish_release but needs to be done outside of docker.
   cross_build:

Makefile

@@ -180,6 +180,24 @@ test: check-git test-deps
 	@echo ">> running all tests. Do export THANOS_SKIP_GCS_TESTS='true' or/and THANOS_SKIP_S3_AWS_TESTS='true' or/and THANOS_SKIP_AZURE_TESTS='true' and/or THANOS_SKIP_SWIFT_TESTS='true' and/or THANOS_SKIP_TENCENT_COS_TESTS='true' if you want to skip e2e tests against real store buckets"
 	THANOS_TEST_PROMETHEUS_VERSIONS="$(PROM_VERSIONS)" THANOS_TEST_ALERTMANAGER_PATH="alertmanager-$(ALERTMANAGER_VERSION)" go test $(shell go list ./... | grep -v /vendor/ | grep -v /benchmark/);
 
+.PHONY: test-only-gcs
+test-only-gcs: export THANOS_SKIP_S3_AWS_TESTS = true
+test-only-gcs: export THANOS_SKIP_AZURE_TESTS = true
+test-only-gcs: export THANOS_SKIP_SWIFT_TESTS = true
+test-only-gcs: export THANOS_SKIP_TENCENT_COS_TESTS = true
+test-only-gcs:
+	@echo ">> Skipping S3 tests"
+	@echo ">> Skipping AZURE tests"
+	@echo ">> Skipping SWIFT tests"
+	@echo ">> Skipping TENCENT tests"
+	$(MAKE) test
+
+.PHONY: test-local
+test-local: export THANOS_SKIP_GCS_TESTS = true
+test-local:
+	@echo ">> Skipping GCE tests"
+	$(MAKE) test-only-gcs
+
 # test-deps installs dependency for e2e tets.
 # It installs current Thanos, supported versions of Prometheus and alertmanager to test against in e2e.
 .PHONY: test-deps
@@ -212,9 +230,10 @@ web: web-pre-process $(HUGO)
 
 .PHONY: lint
 lint: check-git $(GOLANGCILINT)
+	@echo ">> checking formatting"
+	@$(GOLANGCILINT) run -v --disable-all -E goimports ./...
 	@echo ">> linting all of the Go files"
-	@$(GOLANGCILINT) run --disable-all -E goimports ./...
-	@$(GOLANGCILINT) run ./...
+	@$(GOLANGCILINT) run -v ./...
 
 .PHONY: web-serve
 web-serve: web-pre-process $(HUGO)
@@ -242,6 +261,9 @@ $(HUGO):
 $(GOBINDATA):
 	$(call fetch_go_bin_version,github.com/go-bindata/go-bindata/go-bindata,$(GOBINDATA_VERSION))
 
+.PHONY: golangclint
+golangclint: $(GOLANGCILINT)
+
 $(GOLANGCILINT):
 	$(call fetch_go_bin_version,github.com/golangci/golangci-lint/cmd/golangci-lint,$(GOLANGCILINT_VERSION))
 

cmd/thanos/query.go

@@ -399,8 +399,7 @@ func runQuery(
 		}
 
 		ins := extpromhttp.NewInstrumentationMiddleware(reg)
-
-		ui.NewQueryUI(logger, stores, flagsMap).Register(router.WithPrefix(webRoutePrefix), ins)
+		ui.NewQueryUI(logger, reg, stores, flagsMap).Register(router.WithPrefix(webRoutePrefix), ins)
 
 		api := v1.NewAPI(logger, reg, engine, queryableCreator, enableAutodownsampling, enablePartialResponse)
 

cmd/thanos/rule.go

@@ -529,9 +529,9 @@ func runRule(
 
 		ins := extpromhttp.NewInstrumentationMiddleware(reg)
 
-		ui.NewRuleUI(logger, ruleMgrs, alertQueryURL.String(), flagsMap).Register(router.WithPrefix(webRoutePrefix), ins)
+		ui.NewRuleUI(logger, reg, ruleMgrs, alertQueryURL.String(), flagsMap).Register(router.WithPrefix(webRoutePrefix), ins)
 
-		api := v1.NewAPI(logger, ruleMgrs)
+		api := v1.NewAPI(logger, reg, ruleMgrs)
 		api.Register(router.WithPrefix(path.Join(webRoutePrefix, "/api/v1")), tracer, logger, ins)
 
 		mux := http.NewServeMux()

go.mod

@@ -1,5 +1,8 @@
 module github.com/thanos-io/thanos
 
+// TODO(bwplotka): Pin to https://github.com/prometheus/prometheus/pull/5703 until it's merged.
+replace github.com/prometheus/prometheus => github.com/prometheus/prometheus v0.0.0-20190814100250-bd8bf91426aa
+
 require (
 	cloud.google.com/go v0.44.1
 	github.com/Azure/azure-storage-blob-go v0.7.0

pkg/query/api/v1.go

@@ -104,7 +104,9 @@ type API struct {
 	rangeQueryDuration     prometheus.Histogram
 	enableAutodownsampling bool
 	enablePartialResponse  bool
-	now                    func() time.Time
+	reg                    prometheus.Registerer
+
+	now func() time.Time
 }
 
 // NewAPI returns an initialized API type.
@@ -143,6 +145,7 @@ func NewAPI(
 		rangeQueryDuration:     rangeQueryDuration,
 		enableAutodownsampling: enableAutodownsampling,
 		enablePartialResponse:  enablePartialResponse,
+		reg:                    reg,
 
 		now: time.Now,
 	}

pkg/receive/handler.go

@@ -23,10 +23,10 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/prometheus/common/route"
+	"github.com/prometheus/prometheus/prompb"
 	promtsdb "github.com/prometheus/prometheus/storage/tsdb"
 	terrors "github.com/prometheus/prometheus/tsdb/errors"
 	"github.com/thanos-io/thanos/pkg/runutil"
-	"github.com/thanos-io/thanos/pkg/store/prompb"
 )
 
 // Options for the web Handler.

pkg/receive/hashring.go

@@ -7,7 +7,7 @@ import (
 	"sort"
 	"sync"
 
-	"github.com/thanos-io/thanos/pkg/store/prompb"
+	"github.com/prometheus/prometheus/prompb"
 
 	"github.com/cespare/xxhash"
 )

pkg/receive/hashring_test.go

@@ -3,7 +3,7 @@ package receive
 import (
 	"testing"
 
-	"github.com/thanos-io/thanos/pkg/store/prompb"
+	"github.com/prometheus/prometheus/prompb"
 )
 
 func TestHash(t *testing.T) {

pkg/receive/writer.go

@@ -3,7 +3,7 @@ package receive
 import (
 	"github.com/go-kit/kit/log"
 	"github.com/pkg/errors"
-	"github.com/thanos-io/thanos/pkg/store/prompb"
+	"github.com/prometheus/prometheus/prompb"
 
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/storage"

pkg/reloader/reloader_test.go

@@ -211,7 +211,7 @@ func TestReloader_RuleApply(t *testing.T) {
 	testutil.Ok(t, ioutil.WriteFile(path.Join(dir2, "rule3-source.yaml"), []byte("rule3"), os.ModePerm))
 	testutil.Ok(t, ioutil.WriteFile(path.Join(dir2, "rule-dir", "rule4.yaml"), []byte("rule4"), os.ModePerm))
 
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	g := sync.WaitGroup{}
 	g.Add(1)
 	go func() {

pkg/rule/api/v1.go

@@ -6,6 +6,8 @@ import (
 	"time"
 
 	"github.com/NYTimes/gziphandler"
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/go-kit/kit/log"
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/prometheus/common/route"
@@ -22,16 +24,19 @@ type API struct {
 	logger        log.Logger
 	now           func() time.Time
 	ruleRetriever RulesRetriever
+	reg           prometheus.Registerer
 }
 
 func NewAPI(
 	logger log.Logger,
+	reg prometheus.Registerer,
 	ruleRetriever RulesRetriever,
 ) *API {
 	return &API{
 		logger:        logger,
 		now:           time.Now,
 		ruleRetriever: ruleRetriever,
+		reg:           reg,
 	}
 }
 

pkg/rule/api/v1_test.go

@@ -12,15 +12,15 @@ import (
 	"testing"
 	"time"
 
-	"github.com/prometheus/common/model"
-	"github.com/prometheus/prometheus/storage"
-	"github.com/prometheus/prometheus/storage/tsdb"
-
 	"github.com/go-kit/kit/log"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/model"
 	"github.com/prometheus/common/route"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/rules"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/prometheus/prometheus/storage/tsdb"
 	qapi "github.com/thanos-io/thanos/pkg/query/api"
 	thanosrule "github.com/thanos-io/thanos/pkg/rule"
 )
@@ -164,6 +164,7 @@ func TestEndpoints(t *testing.T) {
 		algr.RuleGroups()
 		api := NewAPI(
 			nil,
+			prometheus.DefaultRegisterer,
 			algr,
 		)
 		testEndpoints(t, api)

pkg/rule/rule.go

@@ -175,6 +175,7 @@ func (m *Managers) Update(dataDir string, evalInterval time.Duration, files []st
 			continue
 		}
 		// We add external labels in `pkg/alert.Queue`.
+		// TODO(bwplotka): Investigate if we should put ext labels here or not.
 		if err := updater.Update(evalInterval, fs, nil); err != nil {
 			errs = append(errs, err)
 			continue