[integration] Add query fanout docker integration test (#1652)

scripts/docker-integration-tests/common.sh

@@ -48,12 +48,17 @@ function setup_single_m3db_node {
 }
 
 function wait_for_db_init {
+  local dbnode_host=${DBNODE_HOST:-dbnode01}
+  local dbnode_port=${DBNODE_PORT:-9000}
+  local dbnode_health_port=${DBNODE_HEALTH_PORT:-9002}
+  local coordinator_port=${COORDINATOR_PORT:-7201}
+
   echo "Wait for API to be available"
   ATTEMPTS=10 TIMEOUT=2 retry_with_backoff  \
-    '[ "$(curl -sSf 0.0.0.0:7201/api/v1/namespace | jq ".namespaces | length")" == "0" ]'
+    '[ "$(curl -sSf 0.0.0.0:'"${coordinator_port}"'/api/v1/namespace | jq ".namespaces | length")" == "0" ]'
 
   echo "Adding placement and agg namespace"
-  curl -vvvsSf -X POST 0.0.0.0:7201/api/v1/database/create -d '{
+  curl -vvvsSf -X POST 0.0.0.0:${coordinator_port}/api/v1/database/create -d '{
     "type": "cluster",
     "namespaceName": "agg",
     "retentionTime": "6h",
@@ -64,32 +69,32 @@ function wait_for_db_init {
           "isolation_group": "rack-a",
           "zone": "embedded",
           "weight": 1024,
-          "address": "dbnode01",
-          "port": 9000
+          "address": "'"${dbnode_host}"'",
+          "port": '"${dbnode_port}"'
       }
     ]
   }'
 
   echo "Wait until placement is init'd"
   ATTEMPTS=4 TIMEOUT=1 retry_with_backoff  \
-    '[ "$(curl -sSf 0.0.0.0:7201/api/v1/placement | jq .placement.instances.m3db_local.id)" == \"m3db_local\" ]'
+    '[ "$(curl -sSf 0.0.0.0:'"${coordinator_port}"'/api/v1/placement | jq .placement.instances.m3db_local.id)" == \"m3db_local\" ]'
 
   echo "Wait until agg namespace is init'd"
   ATTEMPTS=4 TIMEOUT=1 retry_with_backoff  \
-    '[ "$(curl -sSf 0.0.0.0:7201/api/v1/namespace | jq .registry.namespaces.agg.indexOptions.enabled)" == true ]'
+    '[ "$(curl -sSf 0.0.0.0:'"${coordinator_port}"'/api/v1/namespace | jq .registry.namespaces.agg.indexOptions.enabled)" == true ]'
 
   echo "Adding unagg namespace"
-  curl -vvvsSf -X POST 0.0.0.0:7201/api/v1/database/namespace/create -d '{
+  curl -vvvsSf -X POST 0.0.0.0:${coordinator_port}/api/v1/database/namespace/create -d '{
     "namespaceName": "unagg",
     "retentionTime": "6h"
   }'
 
   echo "Wait until unagg namespace is init'd"
   ATTEMPTS=4 TIMEOUT=1 retry_with_backoff  \
-    '[ "$(curl -sSf 0.0.0.0:7201/api/v1/namespace | jq .registry.namespaces.unagg.indexOptions.enabled)" == true ]'
+    '[ "$(curl -sSf 0.0.0.0:'"${coordinator_port}"'/api/v1/namespace | jq .registry.namespaces.unagg.indexOptions.enabled)" == true ]'
 
   echo "Wait until bootstrapped"
   ATTEMPTS=10 TIMEOUT=2 retry_with_backoff  \
-    '[ "$(curl -sSf 0.0.0.0:9002/health | jq .bootstrapped)" == true ]'
+    '[ "$(curl -sSf 0.0.0.0:'"${dbnode_health_port}"'/health | jq .bootstrapped)" == true ]'
 }
 

scripts/docker-integration-tests/query_fanout/docker-compose.yml

@@ -0,0 +1,48 @@
+version: "3.5"
+services:
+  dbnode-cluster-a:
+    expose:
+      - "9000-9004"
+      - "2379-2380"
+    ports:
+      - "0.0.0.0:9000-9004:9000-9004"
+      - "0.0.0.0:2379-2380:2379-2380"
+    networks:
+      - backend
+    image: "m3dbnode_integration:${REVISION}"
+  coordinator-cluster-a:
+    expose:
+      - "7201"
+      - "7203"
+    ports:
+      - "0.0.0.0:7201:7201"
+      - "0.0.0.0:7203:7203"
+    networks:
+      - backend
+    image: "m3coordinator_integration:${REVISION}"
+    volumes:
+      - "./m3coordinator-cluster-a.yml:/etc/m3coordinator/m3coordinator.yml"
+  dbnode-cluster-b:
+    expose:
+      - "9000-9004"
+      - "2379-2380"
+    ports:
+      - "0.0.0.0:19000-19004:9000-9004"
+      - "0.0.0.0:12379-12380:2379-2380"
+    networks:
+      - backend
+    image: "m3dbnode_integration:${REVISION}"
+  coordinator-cluster-b:
+    expose:
+      - "7201"
+      - "7203"
+    ports:
+      - "0.0.0.0:17201:7201"
+      - "0.0.0.0:17203:7203"
+    networks:
+      - backend
+    image: "m3coordinator_integration:${REVISION}"
+    volumes:
+      - "./m3coordinator-cluster-b.yml:/etc/m3coordinator/m3coordinator.yml"
+networks:
+  backend:

scripts/docker-integration-tests/query_fanout/m3coordinator-cluster-a.yml

@@ -0,0 +1,45 @@
+listenAddress:
+  type: "config"
+  value: "0.0.0.0:7201"
+
+metrics:
+  scope:
+    prefix: "coordinator"
+  prometheus:
+    handlerPath: /metrics
+    listenAddress: 0.0.0.0:7203 # until https://github.com/m3db/m3/issues/682 is resolved
+  sanitization: prometheus
+  samplingRate: 1.0
+  extended: none
+
+# Fanout queries to remote clusters
+rpc:
+  enabled: true
+  listenAddress: "0.0.0.0:7202"
+  remoteListenAddresses: ["coordinator-cluster-b:7202"]
+
+clusters:
+  - namespaces:
+      - namespace: agg
+        type: aggregated
+        retention: 10h
+        resolution: 15s
+      - namespace: unagg
+        type: unaggregated
+        retention: 10m
+    client:
+      config:
+        service:
+          env: default_env
+          zone: embedded
+          service: m3db
+          cacheDir: /var/lib/m3kv
+          etcdClusters:
+            - zone: embedded
+              endpoints:
+                - dbnode-cluster-a:2379
+      writeConsistencyLevel: majority
+      readConsistencyLevel: unstrict_majority
+
+tagOptions:
+  idScheme: quoted

scripts/docker-integration-tests/query_fanout/m3coordinator-cluster-b.yml

@@ -0,0 +1,45 @@
+listenAddress:
+  type: "config"
+  value: "0.0.0.0:7201"
+
+metrics:
+  scope:
+    prefix: "coordinator"
+  prometheus:
+    handlerPath: /metrics
+    listenAddress: 0.0.0.0:7203 # until https://github.com/m3db/m3/issues/682 is resolved
+  sanitization: prometheus
+  samplingRate: 1.0
+  extended: none
+
+# Fanout queries to remote clusters
+rpc:
+  enabled: true
+  listenAddress: "0.0.0.0:7202"
+  remoteListenAddresses: ["coordinator-cluster-a:7202"]
+
+clusters:
+  - namespaces:
+      - namespace: agg
+        type: aggregated
+        retention: 10h
+        resolution: 15s
+      - namespace: unagg
+        type: unaggregated
+        retention: 10m
+    client:
+      config:
+        service:
+          env: default_env
+          zone: embedded
+          service: m3db
+          cacheDir: /var/lib/m3kv
+          etcdClusters:
+            - zone: embedded
+              endpoints:
+                - dbnode-cluster-b:2379
+      writeConsistencyLevel: majority
+      readConsistencyLevel: unstrict_majority
+
+tagOptions:
+  idScheme: quoted

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

@@ -0,0 +1,92 @@
+#!/usr/bin/env bash
+
+set -xe
+
+source $GOPATH/src/github.com/m3db/m3/scripts/docker-integration-tests/common.sh
+REVISION=$(git rev-parse HEAD)
+COMPOSE_FILE=$GOPATH/src/github.com/m3db/m3/scripts/docker-integration-tests/query_fanout/docker-compose.yml
+export REVISION
+
+echo "Run m3dbnode and m3coordinator containers"
+docker-compose -f ${COMPOSE_FILE} up -d dbnode-cluster-a
+docker-compose -f ${COMPOSE_FILE} up -d coordinator-cluster-a
+
+docker-compose -f ${COMPOSE_FILE} up -d dbnode-cluster-b
+docker-compose -f ${COMPOSE_FILE} up -d coordinator-cluster-b
+
+# think of this as a defer func() in golang
+function defer {
+  docker-compose -f ${COMPOSE_FILE} down || echo "unable to shutdown containers" # CI fails to stop all containers sometimes
+}
+trap defer EXIT
+
+DBNODE_HOST=dbnode-cluster-a DBDNODE_PORT=9000 DBNODE_HEALTH_PORT=9002 COORDINATOR_PORT=7201 \
+ setup_single_m3db_node
+
+DBNODE_HOST=dbnode-cluster-b DBDNODE_PORT=19000 DBNODE_HEALTH_PORT=19002 COORDINATOR_PORT=17201 \
+ setup_single_m3db_node
+
+echo "Write data to cluster a"
+curl -vvvsS -X POST 0.0.0.0:9003/writetagged -d '{
+  "namespace": "unagg",
+  "id": "{__name__=\"test_metric\",cluster=\"cluster-a\",endpoint=\"/request\"}",
+  "tags": [
+    {
+      "name": "__name__",
+      "value": "test_metric"
+    },
+    {
+      "name": "cluster",
+      "value": "cluster-a"
+    },
+    {
+      "name": "endpoint",
+      "value": "/request"
+    }
+  ],
+  "datapoint": {
+    "timestamp":'"$(date +"%s")"',
+    "value": 42.123456789
+  }
+}'
+
+echo "Write 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\"}",
+  "tags": [
+    {
+      "name": "__name__",
+      "value": "test_metric"
+    },
+    {
+      "name": "cluster",
+      "value": "cluster-b"
+    },
+    {
+      "name": "endpoint",
+      "value": "/request"
+    }
+  ],
+  "datapoint": {
+    "timestamp":'"$(date +"%s")"',
+    "value": 42.123456789
+  }
+}'
+
+
+function read {
+  RESPONSE=$(curl "http://0.0.0.0:7201/api/v1/query?query=test_metric")
+  ACTUAL=$(echo $RESPONSE | jq .data.result[].metric.cluster)
+  test "$(echo $ACTUAL)" = '"cluster-a" "cluster-b"'
+}
+
+ATTEMPTS=5 TIMEOUT=1 retry_with_backoff read
+
+function read_sum {
+  RESPONSE=$(curl "http://0.0.0.0:7201/api/v1/query?query=sum(test_metric)")
+  ACTUAL=$(echo $RESPONSE | jq .data.result[].value[1])
+  test $ACTUAL = '"84.246913578"'
+}
+
+ATTEMPTS=5 TIMEOUT=1 retry_with_backoff read_sum

src/query/block/column.go

@@ -285,7 +285,6 @@ func (m *columnBlockSeriesIter) Err() error {
 }
 
 func (m *columnBlockSeriesIter) Next() bool {
-	fmt.Printf("THIS iter4 next\n")
 	m.idx++
 	next := m.idx < m.SeriesCount()
 	if !next {