[query] Adding complete tags to remote server with integration test

Makefile

@@ -227,6 +227,7 @@ docker-integration-test:
 	@./scripts/docker-integration-tests/prometheus/test.sh
 	@./scripts/docker-integration-tests/carbon/test.sh
 	@./scripts/docker-integration-tests/aggregator/test.sh
+	@./scripts/docker-integration-tests/query_fanout/test.sh
 
 .PHONY: site-build
 site-build:

scripts/docker-integration-tests/query_fanout/test.sh

@@ -74,7 +74,6 @@ curl -vvvsS -X POST 0.0.0.0:19003/writetagged -d '{
   }
 }'
 
-
 function read {
   RESPONSE=$(curl "http://0.0.0.0:7201/api/v1/query?query=test_metric")
   ACTUAL=$(echo $RESPONSE | jq .data.result[].metric.cluster)
@@ -90,3 +89,67 @@ function read_sum {
 }
 
 ATTEMPTS=5 TIMEOUT=1 retry_with_backoff read_sum
+
+echo "Write local tagged data to cluster a"
+curl -vvvsS -X POST 0.0.0.0:19003/writetagged -d '{
+  "namespace": "unagg",
+  "id": "{__name__=\"test_metric\",cluster=\"cluster-b\",endpoint=\"/request\",local-only=\"local\"}",
+  "tags": [
+    {
+      "name": "__name__",
+      "value": "test_metric"
+    },
+    {
+      "name": "cluster",
+      "value": "cluster-b"
+    },
+    {
+      "name": "endpoint",
+      "value": "/request"
+    },
+    {
+      "name": "local-only",
+      "value": "local"
+    }
+  ],
+  "datapoint": {
+    "timestamp":'"$(date +"%s")"',
+    "value": 42.123456789
+  }
+}'
+
+echo "Write remote tagged data to cluster b"
+curl -vvvsS -X POST 0.0.0.0:19003/writetagged -d '{
+  "namespace": "unagg",
+  "id": "{__name__=\"test_metric\",cluster=\"cluster-b\",endpoint=\"/request\",remote-only=\"remote\"}",
+  "tags": [
+    {
+      "name": "__name__",
+      "value": "test_metric"
+    },
+    {
+      "name": "cluster",
+      "value": "cluster-b"
+    },
+    {
+      "name": "endpoint",
+      "value": "/request"
+    },
+    {
+      "name": "remote-only",
+      "value": "remote"
+    }
+  ],
+  "datapoint": {
+    "timestamp":'"$(date +"%s")"',
+    "value": 42.123456789
+  }
+}'
+
+function complete_tags {
+  RESPONSE=$(curl "http://0.0.0.0:7201/api/v1/labels")
+  ACTUAL=$(echo $RESPONSE | jq .data[])
+  test "$(echo $ACTUAL)" = '"__name__" "cluster" "endpoint" "local-only" "remote-only"'
+}
+
+ATTEMPTS=5 TIMEOUT=1 retry_with_backoff complete_tags

scripts/docker-integration-tests/setup.sh

@@ -5,7 +5,7 @@ set -xe
 # expected to be run from root of repository
 cd $GOPATH/src/github.com/m3db/m3
 
-SERVICES=(m3dbnode m3coordinator m3aggregator)
+SERVICES=(  m3coordinator  )
 REVISION=$(git rev-parse HEAD)
 if [[ $SKIP_CLEAN != "true" ]]; then
   make clean
@@ -23,9 +23,7 @@ echo "building docker images"
 
 for svc in ${SERVICES[@]}; do
   # only build if image doesn't exist
-  if [[ "$(docker images -q ${svc}_integration:${REVISION} 2> /dev/null)" == "" ]]; then
     echo "creating image for $svc"
     make ${svc}-linux-amd64
     docker build -t "${svc}_integration:${REVISION}" -f ./scripts/docker-integration-tests/${svc}.Dockerfile ./bin
-  fi
 done

src/query/generated/proto/rpcpb/query.proto

@@ -135,6 +135,8 @@ enum CompleteTagsType {
 message CompleteTagsRequestOptions {
 	CompleteTagsType type         = 1;
 	repeated bytes filterNameTags = 2;
+	int64 start                   = 3;
+	int64 end                     = 4;
 }
 
 message CompleteTagsRequest {

src/query/server/server_test.go

@@ -232,7 +232,10 @@ func (s *queryServer) CompleteTags(
 }
 
 func TestGRPCBackend(t *testing.T) {
-	var grpcConfigYAML = `
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	addr := lis.Addr().String()
+	var grpcConfigYAML = fmt.Sprintf(`
 listenAddress:
   type: "config"
   value: "127.0.0.1:17221"
@@ -249,7 +252,7 @@ metrics:
 
 rpc:
   remoteListenAddresses:
-    - "127.0.0.1:17222"
+    - "%s"
 
 backend: grpc
 
@@ -268,17 +271,11 @@ writeWorkerPoolPolicy:
   size: 100
   shards: 1000
   killProbability: 0.3
-`
-
-	// TODO(arnikola): REVERT
-	t.SkipNow()
+`, addr)
 
 	ctrl := gomock.NewController(xtest.Reporter{T: t})
 	defer ctrl.Finish()
 
-	port := "127.0.0.1:17222"
-	lis, err := net.Listen("tcp", port)
-	require.NoError(t, err)
 	s := grpc.NewServer()
 	defer s.GracefulStop()
 	qs := &queryServer{}

src/query/storage/fanout/storage.go

@@ -175,9 +175,14 @@ func (s *fanoutStorage) CompleteTags(
 	query *storage.CompleteTagsQuery,
 	options *storage.FetchOptions,
 ) (*storage.CompleteTagsResult, error) {
-	accumulatedTags := storage.NewCompleteTagsResultBuilder(query.CompleteNameOnly)
 	stores := filterCompleteTagsStores(s.stores, s.completeTagsFilter, *query)
-	for _, store := range stores {
+	// short circuit complete tags
+	if len(stores) == 1 {
+		return stores[0].CompleteTags(ctx, query, options)
+	}
+
+	accumulatedTags := storage.NewCompleteTagsResultBuilder(query.CompleteNameOnly)
+	for i, store := range stores {
 		result, err := store.CompleteTags(ctx, query, options)
 		if err != nil {
 			return nil, err

src/query/tsdb/remote/codecs_complete_tags.go

@@ -93,6 +93,80 @@ func encodeCompleteTagsRequest(
 		Options: &rpc.CompleteTagsRequestOptions{
 			Type:           completionType,
 			FilterNameTags: query.FilterNameTags,
+			Start:          fromTime(query.Start),
+			End:            fromTime(query.End),
 		},
 	}, nil
 }
+
+func decodeCompleteTagsRequest(
+	request *rpc.CompleteTagsRequest,
+) (*storage.CompleteTagsQuery, error) {
+	var (
+		opts     = request.GetOptions()
+		matchers = request.GetTagMatchers()
+	)
+
+	completeNameOnly := opts.GetType() == rpc.CompleteTagsType_TAGNAME
+	tagMatchers, err := decodeTagMatchers(matchers)
+	if err != nil {
+		return nil, err
+	}
+
+	return &storage.CompleteTagsQuery{
+		CompleteNameOnly: completeNameOnly,
+		FilterNameTags:   opts.GetFilterNameTags(),
+		TagMatchers:      tagMatchers,
+		Start:            toTime(opts.GetStart()),
+		End:              toTime(opts.GetEnd()),
+	}, nil
+}
+
+func encodeToCompressedCompleteTagsDefaultResult(
+	results *storage.CompleteTagsResult,
+) (*rpc.CompleteTagsResponse, error) {
+	tags := results.CompletedTags
+	values := make([]*rpc.TagValue, 0, len(tags))
+	for _, tag := range tags {
+		values = append(values, &rpc.TagValue{
+			Key:    tag.Name,
+			Values: tag.Values,
+		})
+	}
+
+	return &rpc.CompleteTagsResponse{
+		Value: &rpc.CompleteTagsResponse_Default{
+			Default: &rpc.TagValues{
+				Values: values,
+			},
+		},
+	}, nil
+}
+
+func encodeToCompressedCompleteTagsNameOnlyResult(
+	results *storage.CompleteTagsResult,
+) (*rpc.CompleteTagsResponse, error) {
+	tags := results.CompletedTags
+	names := make([][]byte, 0, len(tags))
+	for _, tag := range tags {
+		names = append(names, tag.Name)
+	}
+
+	return &rpc.CompleteTagsResponse{
+		Value: &rpc.CompleteTagsResponse_NamesOnly{
+			NamesOnly: &rpc.TagNames{
+				Names: names,
+			},
+		},
+	}, nil
+}
+
+func encodeToCompressedCompleteTagsResult(
+	results *storage.CompleteTagsResult,
+) (*rpc.CompleteTagsResponse, error) {
+	if results.CompleteNameOnly {
+		return encodeToCompressedCompleteTagsNameOnlyResult(results)
+	}
+
+	return encodeToCompressedCompleteTagsDefaultResult(results)
+}

src/query/tsdb/remote/server.go

@@ -25,7 +25,6 @@ import (
 	"time"
 
 	"github.com/m3db/m3/src/dbnode/encoding"
-	"github.com/m3db/m3/src/query/errors"
 	rpc "github.com/m3db/m3/src/query/generated/proto/rpcpb"
 	"github.com/m3db/m3/src/query/models"
 	"github.com/m3db/m3/src/query/pools"
@@ -168,5 +167,36 @@ func (s *grpcServer) CompleteTags(
 	message *rpc.CompleteTagsRequest,
 	stream rpc.Query_CompleteTagsServer,
 ) error {
-	return errors.ErrNotImplemented
+	var err error
+
+	ctx := retrieveMetadata(stream.Context())
+	logger := logging.WithContext(ctx)
+	completeTagsQuery, err := decodeCompleteTagsRequest(message)
+	if err != nil {
+		logger.Error("unable to decode complete tags query", zap.Error(err))
+		return err
+	}
+
+	// TODO(r): Allow propagation of limit from RPC request
+	fetchOpts := storage.NewFetchOptions()
+	fetchOpts.Limit = s.queryContextOpts.LimitMaxTimeseries
+
+	results, err := s.storage.CompleteTags(ctx, completeTagsQuery, fetchOpts)
+	if err != nil {
+		logger.Error("unable to complete tags", zap.Error(err))
+		return err
+	}
+
+	response, err := encodeToCompressedCompleteTagsResult(results)
+	if err != nil {
+		logger.Error("unable to encode complete tags result", zap.Error(err))
+		return err
+	}
+
+	err = stream.SendMsg(response)
+	if err != nil {
+		logger.Error("unable to send complete tags result", zap.Error(err))
+	}
+
+	return err
 }