Setting up M3 Aggregator

Introduction

m3aggregator is used to cluster stateful downsampling and rollup of metrics before they are store in M3DB. The M3 Coordinator also performs this role but is not cluster aware. This means metrics will not get aggregated properly if you send metrics in round robin fashion to multiple M3 Coordinators for the same metrics ingestion source (e.g. Prometheus server).

Similar to M3DB, m3aggregator supports clustering and replication by default. This means that metrics are correctly routed to the instance(s) responsible for aggregating each metric and multiple m3aggregator replicas can be configured such that there are no single points of failure for aggregation.

Configuration

Before setting up m3aggregator, make sure that you have at least one M3DB node running and a dedicated m3coordinator setup.

We highly recommend running with at least a replication factor 2 for a m3aggregator deployment. If you run with replication factor 1 then when you restart an aggregator it will temporarily interrupt good the stream of aggregated metrics and there will be some data loss.

Topology

Initializing aggregator topology

You can setup a m3aggregator topology by issuing a request to your coordinator (be sure to use your own hostnames, number of shards and replication factor):

  1. curl -vvvsSf -H "Cluster-Environment-Name: namespace/m3db-cluster-name" -X POST http://m3dbnode-with-embedded-coordinator:7201/api/v1/services/m3aggregator/placement/init -d '{
  2.     "num_shards": 64,
  3.     "replication_factor": 2,
  4.     "instances": [
  5.         {
  6.             "id": "m3aggregator01:6000",
  7.             "isolation_group": "availability-zone-a",
  8.             "zone": "embedded",
  9.             "weight": 100,
  10.             "endpoint": "m3aggregator01:6000",
  11.             "hostname": "m3aggregator01",
  12.             "port": 6000
  13.         },
  14.         {
  15.             "id": "m3aggregator02:6000",
  16.             "isolation_group": "availability-zone-b",
  17.             "zone": "embedded",
  18.             "weight": 100,
  19.             "endpoint": "m3aggregator02:6000",
  20.             "hostname": "m3aggregator02",
  21.             "port": 6000
  22.         }
  23.     ]
  24. }'

Initializing m3msg topic for m3aggregator to receive from m3coordinators to aggregate metrics

Now we must setup a topic for the m3aggregator to receive unaggregated metrics from m3coordinator instances:

  1. curl -vvvsSf -H "Cluster-Environment-Name: namespace/m3db-cluster-name" -H "Topic-Name: aggregator_ingest" -X POST http://m3dbnode-with-embedded-coordinator:7201/api/v1/topic/init -d '{
  2.     "numberOfShards": 64
  3. }'

Add m3aggregagtor consumer group to ingest topic

Add the m3aggregator placement to receive traffic from the topic (make sure to set message TTL to match your desired maximum in memory retry message buffer):

  1. curl -vvvsSf -H "Cluster-Environment-Name: namespace/m3db-cluster-name" -H "Topic-Name: aggregator_ingest" -X POST http://m3dbnode-with-embedded-coordinator:7201/api/v1/topic -d '{
  2.   "consumerService": {
  3.     "serviceId": {
  4.       "name": "m3aggregator",
  5.       "environment": "namespace/m3db-cluster-name",
  6.       "zone": "embedded"
  7.     },
  8.     "consumptionType": "REPLICATED",
  9.     "messageTtlNanos": "300000000000"
  10.   }
  11. }'

Note: 300000000000 nanoseconds is a TTL of 5 minutes for messages to rebuffer for retry.

Initializing m3msg topic for m3coordinator to receive from m3aggregator to write to M3DB

Now we must setup a topic for the m3coordinator to receive aggregated metrics from m3aggregator instances to write to M3DB:

  1. curl -vvvsSf -H "Cluster-Environment-Name: namespace/m3db-cluster-name" -H "Topic-Name: aggregated_metrics" -X POST http://m3dbnode-with-embedded-coordinator:7201/api/v1/topic/init -d '{
  2.     "numberOfShards": 64
  3. }'

Initializing m3coordinator topology

Then m3coordinator instances need to be configured to receive traffic for this topic (note ingest at port 7507 must match the configured port for your m3coordinator ingest server, see config at bottom of this guide):

  1. curl -vvvsSf -H "Cluster-Environment-Name: namespace/m3db-cluster-name" -X POST http://m3dbnode-with-embedded-coordinator:7201/api/v1/services/m3coordinator/placement/init -d '{
  2.     "instances": [
  3.         {
  4.             "id": "m3coordinator01",
  5.             "zone": "embedded",
  6.             "endpoint": "m3coordinator01:7507",
  7.             "hostname": "m3coordinator01",
  8.             "port": 7507
  9.         }
  10.     ]
  11. }'

Note: When you add or remove m3coordinator instances they must be added to this placement.

Add m3coordinator consumer group to outbound topic

Add the m3coordinator placement to receive traffic from the topic (make sure to set message TTL to match your desired maximum in memory retry message buffer):

  1. curl -vvvsSf -H "Cluster-Environment-Name: namespace/m3db-cluster-name" -H "Topic-Name: aggregated_metrics" -X POST http://m3dbnode-with-embedded-coordinator:7201/api/v1/topic -d '{
  2.   "consumerService": {
  3.     "serviceId": {
  4.       "name": "m3coordinator",
  5.       "environment": "namespace/m3db-cluster-name",
  6.       "zone": "embedded"
  7.     },
  8.     "consumptionType": "SHARED",
  9.     "messageTtlNanos": "300000000000"
  10.   }
  11. }'

Note: 300000000000 nanoseconds is a TTL of 5 minutes for messages to rebuffer for retry.

Running

Dedicated Coordinator

Metrics will still arrive at the m3coordinator, they simply need to be forwarded to an m3aggregator. The m3coordinator then also needs to receive metrics that have been aggregated from the m3aggregator and store them in M3DB, so running an ingestion server should be configured.

Here is the config you should add to your m3coordinator:

  1. # This is for sending metrics to the remote m3aggregators
  2. downsample:
  3.   remoteAggregator:
  4.     client:
  5.       type: m3msg
  6.       m3msg:
  7.         producer:
  8.           writer:
  9.             topicName: aggregator_ingest
  10.             topicServiceOverride:
  11.               zone: embedded
  12.               environment: namespace/m3db-cluster-name
  13.             placement:
  14.               isStaged: true
  15.             placementServiceOverride:
  16.               namespaces:
  17.                 placement: /placement
  18.             connection:
  19.               numConnections: 4
  20.             messagePool:
  21.               size: 16384
  22.               watermark:
  23.                 low: 0.2
  24.                 high: 0.5
  26. # This is for configuring the ingestion server that will receive metrics from the m3aggregators on port 7507
  27. ingest:
  28.   ingester:
  29.     workerPoolSize: 10000
  30.     opPool:
  31.       size: 10000
  32.     retry:
  33.       maxRetries: 3
  34.       jitter: true
  35.     logSampleRate: 0.01
  36.   m3msg:
  37.     server:
  38.       listenAddress: "0.0.0.0:7507"
  39.       retry:
  40.         maxBackoff: 10s
  41.         jitter: true

M3 Aggregator

You can run m3aggregator by either building and running the binary yourself:

  1. make m3aggregator
  2. ./bin/m3aggregator -f ./src/aggregator/config/m3aggregator.yml

Or you can run it with Docker using the Docker file located at docker/m3aggregator/Dockerfile or the publicly provided image quay.io/m3db/m3aggregator:latest.

You can use a config like so, making note of the topics used such as aggregator_ingest and aggregated_metrics and the corresponding environment namespace/m3db-cluster-name:

  1. logging:
  2.   level: info
  4. metrics:
  5.   scope:
  6.     prefix: m3aggregator
  7.   prometheus:
  8.     onError: none
  9.     handlerPath: /metrics
  10.     listenAddress: 0.0.0.0:6002
  11.     timerType: histogram
  12.   sanitization: prometheus
  13.   samplingRate: 1.0
  14.   extended: none
  16. m3msg:
  17.   server:
  18.     listenAddress: 0.0.0.0:6000
  19.     retry:
  20.       maxBackoff: 10s
  21.       jitter: true
  22.   consumer:
  23.     messagePool:
  24.       size: 16384
  25.       watermark:
  26.         low: 0.2
  27.         high: 0.5
  29. http:
  30.   listenAddress: 0.0.0.0:6001
  31.   readTimeout: 60s
  32.   writeTimeout: 60s
  34. kvClient:
  35.   etcd:
  36.     env: namespace/m3db-cluster-name
  37.     zone: embedded
  38.     service: m3aggregator
  39.     cacheDir: /var/lib/m3kv
  40.     etcdClusters:
  41.       - zone: embedded
  42.         endpoints:
  43.           - dbnode01:2379
  45. runtimeOptions:
  46.   kvConfig:
  47.     environment: namespace/m3db-cluster-name
  48.     zone: embedded
  49.   writeValuesPerMetricLimitPerSecondKey: write-values-per-metric-limit-per-second
  50.   writeValuesPerMetricLimitPerSecond: 0
  51.   writeNewMetricLimitClusterPerSecondKey: write-new-metric-limit-cluster-per-second
  52.   writeNewMetricLimitClusterPerSecond: 0
  53.   writeNewMetricNoLimitWarmupDuration: 0
  55. aggregator:
  56.   hostID:
  57.     resolver: environment
  58.     envVarName: M3AGGREGATOR_HOST_ID
  59.   instanceID:
  60.     type: host_id
  61.   verboseErrors: true
  62.   metricPrefix: ""
  63.   counterPrefix: ""
  64.   timerPrefix: ""
  65.   gaugePrefix: ""
  66.   aggregationTypes:
  67.     counterTransformFnType: empty
  68.     timerTransformFnType: suffix
  69.     gaugeTransformFnType: empty
  70.     aggregationTypesPool:
  71.       size: 1024
  72.     quantilesPool:
  73.       buckets:
  74.         - count: 256
  75.           capacity: 4
  76.         - count: 128
  77.           capacity: 8
  78.   stream:
  79.     eps: 0.001
  80.     capacity: 32
  81.     streamPool:
  82.       size: 4096
  83.     samplePool:
  84.       size: 4096
  85.     floatsPool:
  86.       buckets:
  87.         - count: 4096
  88.           capacity: 16
  89.         - count: 2048
  90.           capacity: 32
  91.         - count: 1024
  92.           capacity: 64
  93.   client:
  94.     type: m3msg
  95.     m3msg:
  96.       producer:
  97.         writer:
  98.           topicName: aggregator_ingest
  99.           topicServiceOverride:
  100.             zone: embedded
  101.             environment: namespace/m3db-cluster-name
  102.           placement:
  103.             isStaged: true
  104.           placementServiceOverride:
  105.             namespaces:
  106.               placement: /placement
  107.           messagePool:
  108.             size: 16384
  109.             watermark:
  110.               low: 0.2
  111.               high: 0.5
  112.   placementManager:
  113.     kvConfig:
  114.       namespace: /placement
  115.       environment: namespace/m3db-cluster-name
  116.       zone: embedded
  117.     placementWatcher:
  118.       key: m3aggregator
  119.       initWatchTimeout: 10s
  120.   hashType: murmur32
  121.   bufferDurationBeforeShardCutover: 10m
  122.   bufferDurationAfterShardCutoff: 10m
  123.   bufferDurationForFutureTimedMetric: 10m # Allow test to write into future.
  124.   resignTimeout: 1m
  125.   flushTimesManager:
  126.     kvConfig:
  127.       environment: namespace/m3db-cluster-name
  128.       zone: embedded
  129.     flushTimesKeyFmt: shardset/%d/flush
  130.     flushTimesPersistRetrier:
  131.       initialBackoff: 100ms
  132.       backoffFactor: 2.0
  133.       maxBackoff: 2s
  134.       maxRetries: 3
  135.   electionManager:
  136.     election:
  137.       leaderTimeout: 10s
  138.       resignTimeout: 10s
  139.       ttlSeconds: 10
  140.     serviceID:
  141.       name: m3aggregator
  142.       environment: namespace/m3db-cluster-name
  143.       zone: embedded
  144.     electionKeyFmt: shardset/%d/lock
  145.     campaignRetrier:
  146.       initialBackoff: 100ms
  147.       backoffFactor: 2.0
  148.       maxBackoff: 2s
  149.       forever: true
  150.       jitter: true
  151.     changeRetrier:
  152.       initialBackoff: 100ms
  153.       backoffFactor: 2.0
  154.       maxBackoff: 5s
  155.       forever: true
  156.       jitter: true
  157.     resignRetrier:
  158.       initialBackoff: 100ms
  159.       backoffFactor: 2.0
  160.       maxBackoff: 5s
  161.       forever: true
  162.       jitter: true
  163.     campaignStateCheckInterval: 1s
  164.     shardCutoffCheckOffset: 30s
  165.   flushManager:
  166.     checkEvery: 1s
  167.     jitterEnabled: true
  168.     maxJitters:
  169.       - flushInterval: 5s
  170.         maxJitterPercent: 1.0
  171.       - flushInterval: 10s
  172.         maxJitterPercent: 0.5
  173.       - flushInterval: 1m
  174.         maxJitterPercent: 0.5
  175.       - flushInterval: 10m
  176.         maxJitterPercent: 0.5
  177.       - flushInterval: 1h
  178.         maxJitterPercent: 0.25
  179.     numWorkersPerCPU: 0.5
  180.     flushTimesPersistEvery: 10s
  181.     maxBufferSize: 5m
  182.     forcedFlushWindowSize: 10s
  183.   flush:
  184.     handlers:
  185.       - dynamicBackend:
  186.           name: m3msg
  187.           hashType: murmur32
  188.           producer:
  189.             writer:
  190.               topicName: aggregated_metrics
  191.               topicServiceOverride:
  192.                 zone: embedded
  193.                 environment: namespace/m3db-cluster-name
  194.               messagePool:
  195.                 size: 16384
  196.                 watermark:
  197.                   low: 0.2
  198.                   high: 0.5
  199.   passthrough:
  200.     enabled: true
  201.   forwarding:
  202.     maxConstDelay: 5m # Need to add some buffer window, since timed metrics by default are delayed by 1min.
  203.   entryTTL: 1h
  204.   entryCheckInterval: 10m
  205.   maxTimerBatchSizePerWrite: 140
  206.   defaultStoragePolicies: []
  207.   maxNumCachedSourceSets: 2
  208.   discardNaNAggregatedValues: true
  209.   entryPool:
  210.     size: 4096
  211.   counterElemPool:
  212.     size: 4096
  213.   timerElemPool:
  214.     size: 4096
  215.   gaugeElemPool:
  216.     size: 4096

Usage

Send metrics as usual to your m3coordinator instances in round robin fashion (or any other load balancing strategy), the metrics will be forwarded to the m3aggregator instances, then once aggregated they will be returned to the m3coordinator instances to write to M3DB.