diff --git a/caso/messenger/__init__.py b/caso/messenger/__init__.py
index d1489887..53f31eb8 100644
--- a/caso/messenger/__init__.py
+++ b/caso/messenger/__init__.py
@@ -17,17 +17,78 @@
 """Module containing the base class and manager for the cASO messengers."""
 
 import abc
+import typing
 
 from oslo_config import cfg
 from oslo_log import log
 import six
 
+import caso.record
 from caso import loading
 
 CONF = cfg.CONF
 
 LOG = log.getLogger(__name__)
 
+# Valid record types that can be configured
+VALID_RECORD_TYPES = frozenset(["cloud", "ip", "accelerator", "storage", "energy"])
+
+# Default record types for SSM messenger (records from default extractors)
+# Default extractors are: nova, cinder, neutron
+# nova -> CloudRecord, AcceleratorRecord
+# cinder -> StorageRecord
+# neutron -> IPRecord
+DEFAULT_SSM_RECORD_TYPES = ["cloud", "ip", "accelerator", "storage"]
+
+
+def get_messenger_opts(messenger_name: str) -> typing.List[cfg.Opt]:
+    """Get the configuration options for a specific messenger.
+
+    :param messenger_name: Name of the messenger.
+    :returns: List of configuration options for the messenger.
+    """
+    # SSM messengers have a different default (only records from default extractors)
+    if messenger_name in ("ssm", "ssmv4"):
+        default_record_types = DEFAULT_SSM_RECORD_TYPES
+    else:
+        default_record_types = []
+
+    return [
+        cfg.ListOpt(
+            "record_types",
+            default=default_record_types,
+            help="List of record types to publish to this messenger. "
+            "Valid values are: cloud, ip, accelerator, storage, energy. "
+            "If empty, all record types will be published. "
+            f"Default for {messenger_name}: "
+            f"{default_record_types if default_record_types else 'all record types'}.",
+        ),
+    ]
+
+
+def register_messenger_opts(messenger_names: typing.Optional[typing.List[str]] = None):
+    """Register configuration options for the specified messengers.
+
+    :param messenger_names: List of messenger names to register options for.
+                           If None, registers options for all available messengers.
+    """
+    if messenger_names is None:
+        messenger_names = list(loading.get_available_messenger_names())
+
+    for messenger_name in messenger_names:
+        group_name = f"messenger_{messenger_name}"
+        CONF.register_opts(get_messenger_opts(messenger_name), group=group_name)
+
+
+# Mapping from record type names to record classes
+RECORD_TYPE_MAP: typing.Dict[str, type] = {
+    "cloud": caso.record.CloudRecord,
+    "ip": caso.record.IPRecord,
+    "accelerator": caso.record.AcceleratorRecord,
+    "storage": caso.record.StorageRecord,
+    "energy": caso.record.EnergyRecord,
+}
+
 
 @six.add_metaclass(abc.ABCMeta)
 class BaseMessenger(object):
@@ -38,11 +99,37 @@ def push(self, records):
         """Push the records."""
 
 
+def _filter_records(
+    records: typing.List,
+    record_types: typing.Optional[typing.List[str]],
+) -> typing.List:
+    """Filter records based on allowed record types.
+
+    :param records: List of records to filter.
+    :param record_types: List of allowed record type names. If None or empty,
+                         all records are returned.
+    :returns: Filtered list of records.
+    """
+    if not record_types:
+        return records
+
+    allowed_classes = tuple(
+        RECORD_TYPE_MAP[rt] for rt in record_types if rt in RECORD_TYPE_MAP
+    )
+    if not allowed_classes:
+        return records
+
+    return [r for r in records if isinstance(r, allowed_classes)]
+
+
 class Manager(object):
     """Manager for all cASO messengers."""
 
     def __init__(self):
         """Init the manager with all the configured messengers."""
+        # Register messenger options for all configured messengers
+        register_messenger_opts(CONF.messengers)
+
         try:
             self.mgr = loading.get_enabled_messengers(CONF.messengers)
         except Exception as e:
@@ -50,10 +137,36 @@ def __init__(self):
             LOG.error(e)
             raise e
 
+        # Build mapping of messenger name to allowed record types
+        self.messenger_record_types: typing.Dict[
+            str, typing.Optional[typing.List[str]]
+        ] = {}
+        for messenger_name in CONF.messengers:
+            group_name = f"messenger_{messenger_name}"
+            self.messenger_record_types[messenger_name] = None
+            if hasattr(CONF, group_name):
+                group = getattr(CONF, group_name)
+                if hasattr(group, "record_types"):
+                    record_types = group.record_types
+                    if record_types:
+                        self.messenger_record_types[messenger_name] = list(record_types)
+
     def push_to_all(self, records):
         """Push records to all the configured messengers."""
         try:
-            self.mgr.map_method("push", records)
+            for ext in self.mgr:
+                messenger_name = ext.name
+                record_types = self.messenger_record_types.get(messenger_name)
+                filtered_records = _filter_records(records, record_types)
+                if filtered_records:
+                    ext.obj.push(filtered_records)
+                else:
+                    LOG.debug(
+                        "No records to push to messenger %s "
+                        "after filtering for record types: %s",
+                        messenger_name,
+                        record_types,
+                    )
         except Exception as e:
             # Capture exception so that we can continue working
             LOG.error("Something happeneded when pushing records.")
diff --git a/caso/opts.py b/caso/opts.py
index d8112c93..aa568fba 100644
--- a/caso/opts.py
+++ b/caso/opts.py
@@ -23,14 +23,16 @@
 import caso.extract.openstack.nova
 import caso.extract.prometheus
 import caso.keystone_client
+import caso.loading
 import caso.manager
+import caso.messenger
 import caso.messenger.logstash
 import caso.messenger.ssm
 
 
 def list_opts():
     """Get the list of all configured options."""
-    return [
+    opts = [
         (
             "DEFAULT",
             itertools.chain(
@@ -47,3 +49,10 @@ def list_opts():
         ("prometheus", caso.extract.prometheus.opts),
         ("ssm", caso.messenger.ssm.opts),
     ]
+
+    # Add messenger-specific record_types options for all available messengers
+    for messenger_name in caso.loading.get_available_messenger_names():
+        group_name = f"messenger_{messenger_name}"
+        opts.append((group_name, caso.messenger.get_messenger_opts(messenger_name)))
+
+    return opts
diff --git a/caso/tests/test_messenger.py b/caso/tests/test_messenger.py
new file mode 100644
index 00000000..7c08dd8a
--- /dev/null
+++ b/caso/tests/test_messenger.py
@@ -0,0 +1,189 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2014 Spanish National Research Council (CSIC)
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""Tests for messenger module."""
+
+import caso.messenger
+import caso.record
+
+
+class TestFilterRecords:
+    """Test cases for the _filter_records function."""
+
+    def test_filter_records_returns_all_when_no_filter(
+        self, cloud_record, ip_record, storage_record
+    ):
+        """Test that all records are returned when no filter is specified."""
+        records = [cloud_record, ip_record, storage_record]
+        result = caso.messenger._filter_records(records, None)
+        assert result == records
+
+    def test_filter_records_returns_all_when_empty_filter(
+        self, cloud_record, ip_record, storage_record
+    ):
+        """Test that all records are returned when filter list is empty."""
+        records = [cloud_record, ip_record, storage_record]
+        result = caso.messenger._filter_records(records, [])
+        assert result == records
+
+    def test_filter_records_filters_by_cloud(
+        self, cloud_record, ip_record, storage_record
+    ):
+        """Test filtering for cloud records only."""
+        records = [cloud_record, ip_record, storage_record]
+        result = caso.messenger._filter_records(records, ["cloud"])
+        assert len(result) == 1
+        assert isinstance(result[0], caso.record.CloudRecord)
+
+    def test_filter_records_filters_by_ip(
+        self, cloud_record, ip_record, storage_record
+    ):
+        """Test filtering for IP records only."""
+        records = [cloud_record, ip_record, storage_record]
+        result = caso.messenger._filter_records(records, ["ip"])
+        assert len(result) == 1
+        assert isinstance(result[0], caso.record.IPRecord)
+
+    def test_filter_records_filters_by_storage(
+        self, cloud_record, ip_record, storage_record
+    ):
+        """Test filtering for storage records only."""
+        records = [cloud_record, ip_record, storage_record]
+        result = caso.messenger._filter_records(records, ["storage"])
+        assert len(result) == 1
+        assert isinstance(result[0], caso.record.StorageRecord)
+
+    def test_filter_records_filters_by_multiple_types(
+        self, cloud_record, ip_record, storage_record
+    ):
+        """Test filtering for multiple record types."""
+        records = [cloud_record, ip_record, storage_record]
+        result = caso.messenger._filter_records(records, ["cloud", "ip"])
+        assert len(result) == 2
+        assert any(isinstance(r, caso.record.CloudRecord) for r in result)
+        assert any(isinstance(r, caso.record.IPRecord) for r in result)
+
+    def test_filter_records_handles_accelerator(self, accelerator_record):
+        """Test filtering for accelerator records."""
+        records = [accelerator_record]
+        result = caso.messenger._filter_records(records, ["accelerator"])
+        assert len(result) == 1
+        assert isinstance(result[0], caso.record.AcceleratorRecord)
+
+    def test_filter_records_handles_energy(self, energy_record):
+        """Test filtering for energy records."""
+        records = [energy_record]
+        result = caso.messenger._filter_records(records, ["energy"])
+        assert len(result) == 1
+        assert isinstance(result[0], caso.record.EnergyRecord)
+
+    def test_filter_records_returns_empty_when_no_match(self, cloud_record, ip_record):
+        """Test that empty list is returned when no records match filter."""
+        records = [cloud_record, ip_record]
+        result = caso.messenger._filter_records(records, ["storage", "energy"])
+        assert result == []
+
+    def test_filter_records_ignores_invalid_types(self, cloud_record, ip_record):
+        """Test that invalid record types are ignored in the filter."""
+        records = [cloud_record, ip_record]
+        result = caso.messenger._filter_records(records, ["invalid_type"])
+        # When all filter types are invalid, return all records
+        assert result == records
+
+    def test_filter_records_mixed_valid_invalid_types(
+        self, cloud_record, ip_record, storage_record
+    ):
+        """Test filtering with mix of valid and invalid types."""
+        records = [cloud_record, ip_record, storage_record]
+        result = caso.messenger._filter_records(records, ["cloud", "invalid_type"])
+        assert len(result) == 1
+        assert isinstance(result[0], caso.record.CloudRecord)
+
+
+class TestGetMessengerOpts:
+    """Test cases for the get_messenger_opts function."""
+
+    def test_ssm_messenger_has_default_record_types(self):
+        """Test that SSM messenger has default record types."""
+        opts = caso.messenger.get_messenger_opts("ssm")
+        assert len(opts) == 1
+        opt = opts[0]
+        assert opt.name == "record_types"
+        assert opt.default == caso.messenger.DEFAULT_SSM_RECORD_TYPES
+
+    def test_ssmv4_messenger_has_default_record_types(self):
+        """Test that SSMv4 messenger has default record types."""
+        opts = caso.messenger.get_messenger_opts("ssmv4")
+        assert len(opts) == 1
+        opt = opts[0]
+        assert opt.name == "record_types"
+        assert opt.default == caso.messenger.DEFAULT_SSM_RECORD_TYPES
+
+    def test_noop_messenger_has_empty_default(self):
+        """Test that noop messenger has empty default (all record types)."""
+        opts = caso.messenger.get_messenger_opts("noop")
+        assert len(opts) == 1
+        opt = opts[0]
+        assert opt.name == "record_types"
+        assert opt.default == []
+
+    def test_logstash_messenger_has_empty_default(self):
+        """Test that logstash messenger has empty default (all record types)."""
+        opts = caso.messenger.get_messenger_opts("logstash")
+        assert len(opts) == 1
+        opt = opts[0]
+        assert opt.name == "record_types"
+        assert opt.default == []
+
+
+class TestDefaultSsmRecordTypes:
+    """Test cases for default SSM record types."""
+
+    def test_default_ssm_record_types_includes_cloud(self):
+        """Test that default SSM record types includes cloud."""
+        assert "cloud" in caso.messenger.DEFAULT_SSM_RECORD_TYPES
+
+    def test_default_ssm_record_types_includes_ip(self):
+        """Test that default SSM record types includes ip."""
+        assert "ip" in caso.messenger.DEFAULT_SSM_RECORD_TYPES
+
+    def test_default_ssm_record_types_includes_accelerator(self):
+        """Test that default SSM record types includes accelerator."""
+        assert "accelerator" in caso.messenger.DEFAULT_SSM_RECORD_TYPES
+
+    def test_default_ssm_record_types_includes_storage(self):
+        """Test that default SSM record types includes storage."""
+        assert "storage" in caso.messenger.DEFAULT_SSM_RECORD_TYPES
+
+    def test_default_ssm_record_types_excludes_energy(self):
+        """Test that default SSM record types excludes energy."""
+        assert "energy" not in caso.messenger.DEFAULT_SSM_RECORD_TYPES
+
+
+class TestValidRecordTypes:
+    """Test cases for VALID_RECORD_TYPES constant."""
+
+    def test_valid_record_types_contains_all_types(self):
+        """Test that VALID_RECORD_TYPES contains all expected types."""
+        expected = {"cloud", "ip", "accelerator", "storage", "energy"}
+        assert caso.messenger.VALID_RECORD_TYPES == expected
+
+    def test_record_type_map_matches_valid_record_types(self):
+        """Test that RECORD_TYPE_MAP keys match VALID_RECORD_TYPES."""
+        assert (
+            set(caso.messenger.RECORD_TYPE_MAP.keys())
+            == caso.messenger.VALID_RECORD_TYPES
+        )
diff --git a/doc/source/configuration.rst b/doc/source/configuration.rst
index 75e855b1..7d98047e 100644
--- a/doc/source/configuration.rst
+++ b/doc/source/configuration.rst
@@ -208,6 +208,75 @@ messenger. Available options:
 * ``host`` (default: ``localhost``), host of Logstash server.
 * ``port`` (default: ``5000``), Logstash server port.
 
+Messenger record type filtering
+-------------------------------
+
+Each messenger can be configured to only receive specific record types. This allows
+you to control which types of accounting records are published to each messenger.
+
+The configuration is done via ``[messenger_<name>]`` sections, where ``<name>`` is
+the messenger name (e.g., ``ssm``, ``logstash``, ``noop``). Each section supports
+the following option:
+
+* ``record_types`` (list value). List of record types to publish to this messenger.
+  Valid values are: ``cloud``, ``ip``, ``accelerator``, ``storage``, ``energy``.
+  If empty, all record types will be published.
+
+The following record types are available:
+
+.. list-table:: Record types and their extractors
+   :header-rows: 1
+
+   * - Record Type
+     - Extractor
+     - Description
+
+   * - ``cloud``
+     - ``nova``
+     - VM/compute accounting records
+
+   * - ``ip``
+     - ``neutron``
+     - Public IP usage records
+
+   * - ``accelerator``
+     - ``nova``
+     - Accelerator (GPU) usage records
+
+   * - ``storage``
+     - ``cinder``
+     - Block storage records
+
+   * - ``energy``
+     - ``prometheus``
+     - Energy consumption records
+
+**Default behavior:**
+
+* For ``ssm`` and ``ssmv4`` messengers, the default is to publish only records from
+  the default extractors (``nova``, ``cinder``, ``neutron``), which means:
+  ``cloud``, ``ip``, ``accelerator``, ``storage``. Energy records are not published
+  by default to SSM.
+* For other messengers (``logstash``, ``noop``), all record types are published by
+  default.
+
+**Example configuration:**
+
+To publish only cloud and storage records to SSM::
+
+    [messenger_ssm]
+    record_types = cloud,storage
+
+To publish all record types including energy records to SSM::
+
+    [messenger_ssm]
+    record_types = cloud,ip,accelerator,storage,energy
+
+To publish only energy records to logstash::
+
+    [messenger_logstash]
+    record_types = energy
+
 ``[prometheus]`` section
 ------------------------
 
diff --git a/doc/source/static/caso.conf.sample b/doc/source/static/caso.conf.sample
index 4608a8c2..6afff48d 100644
--- a/doc/source/static/caso.conf.sample
+++ b/doc/source/static/caso.conf.sample
@@ -5,7 +5,8 @@
 #
 
 # List of messengers that will dispatch records. valid values are
-# logstash,noop,ssmv4. You can specify more than one messenger. (list value)
+# logstash,noop,ssm,ssmv4. You can specify more than one messenger. (list
+# value)
 #messengers = noop
 
 # Spool directory. (string value)
@@ -18,7 +19,6 @@
 
 # Extract records but do not push records to SSM. This will not update the last
 # run date. (boolean value)
-# Deprecated group/name - [DEFAULT]/dry_run
 #dry_run = false
 
 # Site name as in GOCDB. (string value)
@@ -27,19 +27,32 @@
 # Service name within the site (string value)
 #service_name = $site_name
 
-# List of projects to extract accounting records from. (list value)
-# Deprecated group/name - [DEFAULT]/tenants
+# List of projects to extract accounting records from. You can use this option,
+# or add 'caso' tag to the project in Keystone. Please refer to the
+# documentation for more details. (list value)
 #projects =
 
-# File containing the VO <-> project mapping as used in Keystone-VOMS. (string
+# Tag used to mark a project in Keystone to be extracted by cASO (string value)
+#caso_tag = caso
+
+# Property key used to get the VO name from the project properties.  (string
 # value)
+#vo_property = VO
+
+# DEPRECATED: File containing the VO <-> project mapping as used in Keystone-
+# VOMS. (string value)
+# This option is deprecated for removal.
+# Its value may be silently ignored in the future.
+# Reason: This option is marked for removal in the next release. Please see the
+# release notes, and migrate your current configuration to use the new project
+# mapping as soon as possible. If you already migrated your configuration,
+# please remove the JSON file to get rid of this message.
 #mapping_file = /etc/caso/voms.json
 
 # Extract record changes until this date. If it is not set, we use now. If a
 # server has ended after this date, it will be included, but the consuption
 # reported will end on this date. If no time zone is specified, UTC will be
 # used. (string value)
-# Deprecated group/name - [DEFAULT]/extract_to
 #extract_to = <None>
 
 # Extract records that have changed after this date. This means that if a
@@ -48,13 +61,12 @@
 # If it is not set, extract records from last run. If it is set to None and
 # last run file is not present, it will extract records from the beginning of
 # time. If no time zone is specified, UTC will be used. (string value)
-# Deprecated group/name - [DEFAULT]/extract_from
 #extract_from = <None>
 
 # Which extractor to use for getting the data. If you do not specify anything,
-# nova will be used. Available choices are frozenset({'neutron', 'nova'}) (list
-# value)
-#extractor = nova
+# nova will be used. Available choices are ['cinder', 'neutron', 'nova',
+# 'prometheus'] (list value)
+#extractor = nova,cinder,neutron
 
 #
 # From oslo.config
@@ -111,11 +123,15 @@
 # Deprecated group/name - [DEFAULT]/logdir
 #log_dir = <None>
 
-# Uses logging handler designed to watch file system. When log file is moved or
-# removed this handler will open a new log file with specified path
+# DEPRECATED: Uses logging handler designed to watch file system. When log file
+# is moved or removed this handler will open a new log file with specified path
 # instantaneously. It makes sense only if log_file option is specified and
 # Linux platform is used. This option is ignored if log_config_append is set.
 # (boolean value)
+# This option is deprecated for removal.
+# Its value may be silently ignored in the future.
+# Reason: This function is known to have bene broken for long time, and depends
+# on the unmaintained library
 #watch_log_file = false
 
 # Use syslog for logging. Existing syslog format is DEPRECATED and will be
@@ -141,9 +157,17 @@
 # set. (boolean value)
 #use_stderr = false
 
-# Log output to Windows Event Log. (boolean value)
+# DEPRECATED: Log output to Windows Event Log. (boolean value)
+# This option is deprecated for removal.
+# Its value may be silently ignored in the future.
+# Reason: Windows support is no longer maintained.
 #use_eventlog = false
 
+# (Optional) Set the 'color' key according to log levels. This option takes
+# effect only when logging to stderr or stdout is used. This option is ignored
+# if log_config_append is set. (boolean value)
+#log_color = false
+
 # The amount of time before the log files are rotated. This option is ignored
 # unless log_rotation_type is set to "interval". (integer value)
 #log_rotate_interval = 1
@@ -216,10 +240,16 @@
 # Maximum number of logged messages per rate_limit_interval. (integer value)
 #rate_limit_burst = 0
 
-# Log level name used by rate limiting: CRITICAL, ERROR, INFO, WARNING, DEBUG
-# or empty string. Logs with level greater or equal to rate_limit_except_level
-# are not filtered. An empty string means that all levels are filtered. (string
-# value)
+# Log level name used by rate limiting. Logs with level greater or equal to
+# rate_limit_except_level are not filtered. An empty string means that all
+# levels are filtered. (string value)
+# Possible values:
+# CRITICAL - <No description provided>
+# ERROR - <No description provided>
+# INFO - <No description provided>
+# WARNING - <No description provided>
+# DEBUG - <No description provided>
+# '' - <No description provided>
 #rate_limit_except_level = CRITICAL
 
 # Enables or disables fatal status of deprecations. (boolean value)
@@ -234,22 +264,18 @@
 
 # Metadata key used to retrieve the accelerator type from the flavor
 # properties. (string value)
-# Deprecated group/name - [DEFAULT]/accelerator_type_key
 #type_key = Accelerator:Type
 
 # Metadata key used to retrieve the accelerator vendor from the flavor
 # properties. (string value)
-# Deprecated group/name - [DEFAULT]/accelerator_vendor_key
 #vendor_key = Accelerator:Vendor
 
 # Metadata key used to retrieve the accelerator model from the flavor
 # properties. (string value)
-# Deprecated group/name - [DEFAULT]/accelerator_model_key
 #model_key = Accelerator:Model
 
 # Metadata key used to retrieve the accelerator number from the flavor
 # properties. (string value)
-# Deprecated group/name - [DEFAULT]/accelerator_number_key
 #number_key = Accelerator:Number
 
 
@@ -261,12 +287,10 @@
 
 # Metadata key used to retrieve the benchmark type from the flavor properties.
 # (string value)
-# Deprecated group/name - [DEFAULT]/benchmark_name_key
 #name_key = accounting:benchmark_type
 
 # Metadata key used to retrieve the benchmark value from the flavor properties.
 # (string value)
-# Deprecated group/name - [DEFAULT]/benchmark_value_key
 #value_key = accounting:benchmark_value
 
 
@@ -374,6 +398,91 @@
 #port = 5000
 
 
+[messenger_logstash]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for logstash: all record types. (list value)
+#record_types =
+
+
+[messenger_noop]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for noop: all record types. (list value)
+#record_types =
+
+
+[messenger_ssm]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for ssm: ['cloud', 'ip', 'accelerator', 'storage']. (list
+# value)
+#record_types = cloud,ip,accelerator,storage
+
+
+[messenger_ssmv4]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for ssmv4: ['cloud', 'ip', 'accelerator', 'storage'].
+# (list value)
+#record_types = cloud,ip,accelerator,storage
+
+
+[prometheus]
+
+#
+# From caso
+#
+
+# Prometheus server endpoint URL. (string value)
+#prometheus_endpoint = http://localhost:9090
+
+# Name of the Prometheus metric to query for energy consumption. (string value)
+#prometheus_metric_name = prometheus_value
+
+# Name of the label that matches the VM UUID in Prometheus metrics. (string
+# value)
+#vm_uuid_label_name = uuid
+
+# List of label filters as key:value pairs to filter the Prometheus metric
+# (e.g., 'type_instance:scaph_process_power_microwatts'). The VM UUID label
+# will be added automatically based on vm_uuid_label_name. (list value)
+#labels = type_instance:scaph_process_power_microwatts
+
+# Frequency between samples in the time series (in seconds). (integer value)
+#prometheus_step_seconds = 30
+
+# Query time range (e.g., '1h', '6h', '24h'). (string value)
+#prometheus_query_range = 1h
+
+# Whether to verify SSL when connecting to Prometheus. (boolean value)
+#prometheus_verify_ssl = true
+
+# CPU normalization factor to apply to energy measurements. (floating point
+# value)
+#cpu_normalization_factor = 1.0
+
+
 [sample_remote_file_source]
 # Example of using a remote_file source
 #
@@ -433,6 +542,10 @@
 # includes the private key. (string value)
 #client_key = <None>
 
+# Timeout is the number of seconds the request will wait for your client to
+# establish a connection to a remote machine call on the socket. (string value)
+#timeout = 60
+
 
 [ssm]
 
diff --git a/etc/caso/caso.conf.sample b/etc/caso/caso.conf.sample
index e382d064..6afff48d 100644
--- a/etc/caso/caso.conf.sample
+++ b/etc/caso/caso.conf.sample
@@ -1,5 +1,73 @@
 [DEFAULT]
 
+#
+# From caso
+#
+
+# List of messengers that will dispatch records. valid values are
+# logstash,noop,ssm,ssmv4. You can specify more than one messenger. (list
+# value)
+#messengers = noop
+
+# Spool directory. (string value)
+#spooldir = /var/spool/caso
+
+# Directory to use for lock files. For security, the specified directory should
+# only be writable by the user running the processes that need locking.
+# Defaults to environment variable CASO_LOCK_PATH or $spooldir (string value)
+#lock_path = $spooldir
+
+# Extract records but do not push records to SSM. This will not update the last
+# run date. (boolean value)
+#dry_run = false
+
+# Site name as in GOCDB. (string value)
+#site_name = <None>
+
+# Service name within the site (string value)
+#service_name = $site_name
+
+# List of projects to extract accounting records from. You can use this option,
+# or add 'caso' tag to the project in Keystone. Please refer to the
+# documentation for more details. (list value)
+#projects =
+
+# Tag used to mark a project in Keystone to be extracted by cASO (string value)
+#caso_tag = caso
+
+# Property key used to get the VO name from the project properties.  (string
+# value)
+#vo_property = VO
+
+# DEPRECATED: File containing the VO <-> project mapping as used in Keystone-
+# VOMS. (string value)
+# This option is deprecated for removal.
+# Its value may be silently ignored in the future.
+# Reason: This option is marked for removal in the next release. Please see the
+# release notes, and migrate your current configuration to use the new project
+# mapping as soon as possible. If you already migrated your configuration,
+# please remove the JSON file to get rid of this message.
+#mapping_file = /etc/caso/voms.json
+
+# Extract record changes until this date. If it is not set, we use now. If a
+# server has ended after this date, it will be included, but the consuption
+# reported will end on this date. If no time zone is specified, UTC will be
+# used. (string value)
+#extract_to = <None>
+
+# Extract records that have changed after this date. This means that if a
+# record has started before this date, and it has changed after this date (i.e.
+# it is still running or it has ended) it will be reported.
+# If it is not set, extract records from last run. If it is set to None and
+# last run file is not present, it will extract records from the beginning of
+# time. If no time zone is specified, UTC will be used. (string value)
+#extract_from = <None>
+
+# Which extractor to use for getting the data. If you do not specify anything,
+# nova will be used. Available choices are ['cinder', 'neutron', 'nova',
+# 'prometheus'] (list value)
+#extractor = nova,cinder,neutron
+
 #
 # From oslo.config
 #
@@ -89,6 +157,12 @@
 # set. (boolean value)
 #use_stderr = false
 
+# DEPRECATED: Log output to Windows Event Log. (boolean value)
+# This option is deprecated for removal.
+# Its value may be silently ignored in the future.
+# Reason: Windows support is no longer maintained.
+#use_eventlog = false
+
 # (Optional) Set the 'color' key according to log levels. This option takes
 # effect only when logging to stderr or stdout is used. This option is ignored
 # if log_config_append is set. (boolean value)
@@ -182,6 +256,233 @@
 #fatal_deprecations = false
 
 
+[accelerator]
+
+#
+# From caso
+#
+
+# Metadata key used to retrieve the accelerator type from the flavor
+# properties. (string value)
+#type_key = Accelerator:Type
+
+# Metadata key used to retrieve the accelerator vendor from the flavor
+# properties. (string value)
+#vendor_key = Accelerator:Vendor
+
+# Metadata key used to retrieve the accelerator model from the flavor
+# properties. (string value)
+#model_key = Accelerator:Model
+
+# Metadata key used to retrieve the accelerator number from the flavor
+# properties. (string value)
+#number_key = Accelerator:Number
+
+
+[benchmark]
+
+#
+# From caso
+#
+
+# Metadata key used to retrieve the benchmark type from the flavor properties.
+# (string value)
+#name_key = accounting:benchmark_type
+
+# Metadata key used to retrieve the benchmark value from the flavor properties.
+# (string value)
+#value_key = accounting:benchmark_value
+
+
+[keystone_auth]
+
+#
+# From caso
+#
+
+# Authentication type to load (string value)
+# Deprecated group/name - [keystone_auth]/auth_plugin
+#auth_type = <None>
+
+# Config Section from which to load plugin specific options (string value)
+#auth_section = <None>
+
+# PEM encoded Certificate Authority to use when verifying HTTPs connections.
+# (string value)
+#cafile = <None>
+
+# PEM encoded client certificate cert file (string value)
+#certfile = <None>
+
+# PEM encoded client certificate key file (string value)
+#keyfile = <None>
+
+# Verify HTTPS connections. (boolean value)
+#insecure = false
+
+# Timeout value for http requests (integer value)
+#timeout = <None>
+
+# Collect per-API call timing information. (boolean value)
+#collect_timing = false
+
+# Log requests to multiple loggers. (boolean value)
+#split_loggers = false
+
+# Authentication URL (string value)
+#auth_url = <None>
+
+# Scope for system operations (string value)
+#system_scope = <None>
+
+# Domain ID to scope to (string value)
+#domain_id = <None>
+
+# Domain name to scope to (string value)
+#domain_name = <None>
+
+# Project ID to scope to (string value)
+# Deprecated group/name - [keystone_auth]/tenant_id
+#project_id = <None>
+
+# Project name to scope to (string value)
+# Deprecated group/name - [keystone_auth]/tenant_name
+#project_name = <None>
+
+# Domain ID containing project (string value)
+#project_domain_id = <None>
+
+# Domain name containing project (string value)
+#project_domain_name = <None>
+
+# ID of the trust to use as a trustee use (string value)
+#trust_id = <None>
+
+# Optional domain ID to use with v3 and v2 parameters. It will be used for both
+# the user and project domain in v3 and ignored in v2 authentication. (string
+# value)
+#default_domain_id = <None>
+
+# Optional domain name to use with v3 API and v2 parameters. It will be used
+# for both the user and project domain in v3 and ignored in v2 authentication.
+# (string value)
+#default_domain_name = <None>
+
+# User id (string value)
+#user_id = <None>
+
+# Username (string value)
+# Deprecated group/name - [keystone_auth]/user_name
+#username = <None>
+
+# User's domain id (string value)
+#user_domain_id = <None>
+
+# User's domain name (string value)
+#user_domain_name = <None>
+
+# User's password (string value)
+#password = <None>
+
+
+[logstash]
+
+#
+# From caso
+#
+
+# Logstash host to send records to. (string value)
+#host = localhost
+
+# Logstash server port. (integer value)
+#port = 5000
+
+
+[messenger_logstash]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for logstash: all record types. (list value)
+#record_types =
+
+
+[messenger_noop]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for noop: all record types. (list value)
+#record_types =
+
+
+[messenger_ssm]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for ssm: ['cloud', 'ip', 'accelerator', 'storage']. (list
+# value)
+#record_types = cloud,ip,accelerator,storage
+
+
+[messenger_ssmv4]
+
+#
+# From caso
+#
+
+# List of record types to publish to this messenger. Valid values are: cloud,
+# ip, accelerator, storage, energy. If empty, all record types will be
+# published. Default for ssmv4: ['cloud', 'ip', 'accelerator', 'storage'].
+# (list value)
+#record_types = cloud,ip,accelerator,storage
+
+
+[prometheus]
+
+#
+# From caso
+#
+
+# Prometheus server endpoint URL. (string value)
+#prometheus_endpoint = http://localhost:9090
+
+# Name of the Prometheus metric to query for energy consumption. (string value)
+#prometheus_metric_name = prometheus_value
+
+# Name of the label that matches the VM UUID in Prometheus metrics. (string
+# value)
+#vm_uuid_label_name = uuid
+
+# List of label filters as key:value pairs to filter the Prometheus metric
+# (e.g., 'type_instance:scaph_process_power_microwatts'). The VM UUID label
+# will be added automatically based on vm_uuid_label_name. (list value)
+#labels = type_instance:scaph_process_power_microwatts
+
+# Frequency between samples in the time series (in seconds). (integer value)
+#prometheus_step_seconds = 30
+
+# Query time range (e.g., '1h', '6h', '24h'). (string value)
+#prometheus_query_range = 1h
+
+# Whether to verify SSL when connecting to Prometheus. (boolean value)
+#prometheus_verify_ssl = true
+
+# CPU normalization factor to apply to energy measurements. (floating point
+# value)
+#cpu_normalization_factor = 1.0
+
+
 [sample_remote_file_source]
 # Example of using a remote_file source
 #
@@ -246,34 +547,14 @@
 #timeout = 60
 
 
-[prometheus]
+[ssm]
 
 #
-# From caso.extract.prometheus
+# From caso
 #
 
-# Prometheus server endpoint URL. (string value)
-#prometheus_endpoint = http://localhost:9090
-
-# Name of the Prometheus metric to query for energy consumption. (string value)
-#prometheus_metric_name = prometheus_value
+# Directory to put the generated SSM records. (string value)
+#output_path = /var/spool/apel/outgoing/openstack
 
-# Name of the label that matches the VM UUID in Prometheus metrics. (string value)
-#vm_uuid_label_name = uuid
-
-# List of label filters as key:value pairs to filter the Prometheus metric (e.g.,
-# 'type_instance:scaph_process_power_microwatts'). The VM UUID label will be added
-# automatically based on vm_uuid_label_name. (list value)
-#labels = type_instance:scaph_process_power_microwatts
-
-# Frequency between samples in the time series (in seconds). (integer value)
-#prometheus_step_seconds = 30
-
-# Query time range (e.g., '1h', '6h', '24h'). (string value)
-#prometheus_query_range = 1h
-
-# Whether to verify SSL when connecting to Prometheus. (boolean value)
-#prometheus_verify_ssl = true
-
-# CPU normalization factor to apply to energy measurements. (floating point value)
-#cpu_normalization_factor = 1.0
+# Maximum number of records to send per message (integer value)
+#max_size = 100
diff --git a/releasenotes/notes/messenger-record-type-filtering-c1e0f230eea04cf7.yaml b/releasenotes/notes/messenger-record-type-filtering-c1e0f230eea04cf7.yaml
new file mode 100644
index 00000000..71e3db53
--- /dev/null
+++ b/releasenotes/notes/messenger-record-type-filtering-c1e0f230eea04cf7.yaml
@@ -0,0 +1,23 @@
+---
+features:
+  - |
+    Add per-messenger configuration to control which record types are published.
+    Each messenger can now be configured with a ``record_types`` option in its
+    ``[messenger_<name>]`` section to specify which record types it should receive.
+    Valid record types are: ``cloud``, ``ip``, ``accelerator``, ``storage``, and
+    ``energy``.
+  - |
+    SSM messengers (``ssm``, ``ssmv4``) now default to publishing only records
+    from the default extractors (nova, cinder, neutron): ``cloud``, ``ip``,
+    ``accelerator``, and ``storage``. Energy records from the prometheus
+    extractor are not published to SSM by default.
+  - |
+    Other messengers (``noop``, ``logstash``) default to publishing all record
+    types.
+upgrade:
+  - |
+    To publish energy records to SSM, you must now explicitly add ``energy`` to
+    the ``record_types`` list in the ``[messenger_ssm]`` section::
+
+      [messenger_ssm]
+      record_types = cloud,ip,accelerator,storage,energy