Setting up M3 Aggregator

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.