diff --git a/heka/map.jinja b/heka/map.jinja
index c7de276..300960e 100644
--- a/heka/map.jinja
+++ b/heka/map.jinja
@@ -30,28 +30,40 @@
 {%- endload %}
 {%- set server = salt['grains.filter_by'](server_defaults, merge=salt['pillar.get']('heka:server')) %}
 
-{%- load_yaml as elasticsearch_defaults %}
-default:
-  elasticsearch_port: 9200
-{%- endload %}
-{% set log_collector = salt['grains.filter_by'](elasticsearch_defaults, merge=salt['pillar.get']('heka:log_collector')) %}
+{% set default_elasticsearch_port = 9200 %}
+{% set default_influxdb_port = 8086 %}
+{% set default_influxdb_time_precision = 'ms' %}
+{% set default_influxdb_timeout = 5000 %}
+{% set default_aggregator_port = 5565 %}
 
-{%- load_yaml as influxdb_defaults %}
-default:
-  influxdb_port: 8086
-  influxdb_time_precision: ms
-  influxdb_timeout: 5000
-{%- endload %}
+{% set log_collector = salt['grains.filter_by']({
+  'default': {
+    'elasticsearch_port': default_elasticsearch_port,
+  }
+}, merge=salt['pillar.get']('heka:log_collector')) %}
 
-{%- load_yaml as metric_collector_defaults %}
-default:
-  aggregator_port: 5565
-{%- endload %}
-{% set metric_collector = salt['grains.filter_by'](
-    metric_collector_defaults,
-    merge=salt['grains.filter_by'](
-        influxdb_defaults,
-        merge=salt['pillar.get']('heka:metric_collector'))) %}
+{% set metric_collector = salt['grains.filter_by']({
+  'default': {
+    'influxdb_port': default_influxdb_port,
+    'influxdb_time_precision': default_influxdb_time_precision,
+    'influxdb_timeout': default_influxdb_timeout,
+    'aggregator_port': default_aggregator_port,
+  }
+}, merge=salt['pillar.get']('heka:metric_collector')) %}
 
-{% set remote_collector = salt['grains.filter_by'](influxdb_defaults, merge=salt['pillar.get']('heka:remote_collector')) %}
-{% set aggregator = salt['grains.filter_by'](influxdb_defaults, merge=salt['pillar.get']('heka:aggregator')) %}
+{% set remote_collector = salt['grains.filter_by']({
+  'default': {
+    'influxdb_port': default_influxdb_port,
+    'influxdb_time_precision': default_influxdb_time_precision,
+    'influxdb_timeout': default_influxdb_timeout,
+    'aggregator_port': default_aggregator_port,
+  }
+}, merge=salt['pillar.get']('heka:remote_collector')) %}
+
+{% set aggregator = salt['grains.filter_by']({
+  'default': {
+    'influxdb_port': default_influxdb_port,
+    'influxdb_time_precision': default_influxdb_time_precision,
+    'influxdb_timeout': default_influxdb_timeout,
+  }
+}, merge=salt['pillar.get']('heka:aggregator')) %}
diff --git a/heka/meta/heka.yml b/heka/meta/heka.yml
index 08b99ea..67254c8 100644
--- a/heka/meta/heka.yml
+++ b/heka/meta/heka.yml
@@ -153,7 +153,7 @@
       module_file: /usr/share/lma_collector/filters/influxdb_accumulator.lua
       module_dir: /usr/share/lma_collector/common;/usr/share/heka/lua_modules
       preserve_data: false
-      message_matcher: "Type =~ /metric$/"
+      message_matcher: "Type == 'heka.sandbox.afd_metric'"
       ticker_interval: 1
       config:
         tag_fields: "deployment_id environment_label tenant_id user_id"
@@ -184,6 +184,13 @@
       encoder: influxdb_encoder
       timeout: {{ remote_collector.influxdb_timeout }}
 {%- endif %}
+{%- if remote_collector.aggregator_host is defined %}
+    aggregator:
+      engine: tcp
+      host: "{{ remote_collector.aggregator_host }}"
+      port: "{{ remote_collector.aggregator_port }}"
+      message_matcher: "Fields[aggregator] == NIL && Type == 'heka.sandbox.afd_metric'"
+{%- endif %}
 aggregator:
   policy:
     # A policy defining that the cluster's status depends on the member with