diff --git a/CHANGELOG.md b/CHANGELOG.md
index aeb028ff..c7047d28 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- A `search_distributor` API to search distributors on defined `Criteria`
+- `Matcher.less_than()` matcher to find the results with fields less than
+  the given value
 - ``Page`` objects may now be directly used as iterables
 
 ### Changed
diff --git a/pubtools/pulplib/_impl/client/client.py b/pubtools/pulplib/_impl/client/client.py
index 8194eeb3..1a5ecf41 100644
--- a/pubtools/pulplib/_impl/client/client.py
+++ b/pubtools/pulplib/_impl/client/client.py
@@ -13,7 +13,7 @@
 
 from ..page import Page
 from ..criteria import Criteria
-from ..model import Repository, MaintenanceReport
+from ..model import Repository, MaintenanceReport, Distributor
 from .search import filters_for_criteria
 from .errors import PulpException
 from .poller import TaskPoller
@@ -168,20 +168,44 @@ def search_repository(self, criteria=None):
                 Each page will contain a collection of
                 :class:`~pubtools.pulplib.Repository` objects.
         """
+        search_options = {"distributors": True}
+        return self._search(
+            Repository, "repositories", criteria=criteria, search_options=search_options
+        )
+
+    def search_distributor(self, criteria=None):
+        """Search the distributors matching the given criteria.
+
+        Args:
+            criteria (:class:`~pubtools.pulplib.Criteria`)
+                A criteria object used for this search.
+                If None, search for all distributors.
+
+        Returns:
+            Future[:class:`~pubtools.pulplib.Page`]
+                A future representing the first page of results.
+
+                Each page will contain a collection of
+                :class:`~pubtools.pulplib.Distributor` objects.
+        """
+        return self._search(Distributor, "distributors", criteria=criteria)
+
+    def _search(self, return_type, resource_type, criteria=None, search_options=None):
         pulp_crit = {
             "skip": 0,
             "limit": self._PAGE_SIZE,
-            "filters": filters_for_criteria(criteria, Repository),
+            "filters": filters_for_criteria(criteria, return_type),
         }
-        search = {"criteria": pulp_crit, "distributors": True}
+        search = {"criteria": pulp_crit}
+        search.update(search_options or {})
 
-        response_f = self._do_search("repositories", search)
+        response_f = self._do_search(resource_type, search)
 
         # When this request is resolved, we'll have the first page of data.
         # We'll need to convert that into a page and also keep going with
         # the search if there's more to be done.
         return f_map(
-            response_f, lambda data: self._handle_page(Repository, search, data)
+            response_f, lambda data: self._handle_page(return_type, search, data)
         )
 
     def get_maintenance_report(self):
diff --git a/pubtools/pulplib/_impl/client/search.py b/pubtools/pulplib/_impl/client/search.py
index ab84e92b..640fb5a4 100644
--- a/pubtools/pulplib/_impl/client/search.py
+++ b/pubtools/pulplib/_impl/client/search.py
@@ -1,3 +1,4 @@
+import datetime
 from pubtools.pulplib._impl.criteria import (
     AndCriteria,
     OrCriteria,
@@ -7,6 +8,7 @@
     EqMatcher,
     InMatcher,
     ExistsMatcher,
+    LessThanMatcher,
 )
 
 from pubtools.pulplib._impl import compat_attr as attr
@@ -21,9 +23,27 @@ def all_subclasses(klass):
     return out
 
 
+def to_mongo_json(value):
+    # Return a value converted to the format expected for a mongo JSON
+    # expression. Only a handful of special types need explicit conversions.
+    if isinstance(value, datetime.datetime):
+        return {"$date": value.strftime("%Y-%m-%dT%H:%M:%SZ")}
+
+    if isinstance(value, (list, tuple)):
+        return [to_mongo_json(elem) for elem in value]
+
+    if isinstance(value, dict):
+        out = {}
+        for (key, val) in value.items():
+            out[key] = to_mongo_json(val)
+        return out
+
+    return value
+
+
 def map_field_for_type(field_name, matcher, type_hint):
     if not type_hint:
-        return (field_name, matcher)
+        return None
 
     attrs_classes = all_subclasses(type_hint)
     attrs_classes = [cls for cls in attrs_classes if attr.has(cls)]
@@ -44,7 +64,7 @@ def map_field_for_type(field_name, matcher, type_hint):
         raise NotImplementedError("Searching on field %s is not supported" % field_name)
 
     # No match => no change, search exactly what was requested
-    return (field_name, matcher)
+    return None
 
 
 def filters_for_criteria(criteria, type_hint=None):
@@ -67,7 +87,9 @@ def filters_for_criteria(criteria, type_hint=None):
         field = criteria._field
         matcher = criteria._matcher
 
-        field, matcher = map_field_for_type(field, matcher, type_hint)
+        mapped = map_field_for_type(field, matcher, type_hint)
+        if mapped:
+            field, matcher = mapped
 
         return {field: field_match(matcher)}
 
@@ -79,12 +101,15 @@ def field_match(to_match):
         return {"$regex": to_match._pattern}
 
     if isinstance(to_match, EqMatcher):
-        return {"$eq": to_match._value}
+        return {"$eq": to_mongo_json(to_match._value)}
 
     if isinstance(to_match, InMatcher):
-        return {"$in": to_match._values}
+        return {"$in": to_mongo_json(to_match._values)}
 
     if isinstance(to_match, ExistsMatcher):
         return {"$exists": True}
 
+    if isinstance(to_match, LessThanMatcher):
+        return {"$lt": to_mongo_json(to_match._value)}
+
     raise TypeError("Not a matcher: %s" % repr(to_match))
diff --git a/pubtools/pulplib/_impl/criteria.py b/pubtools/pulplib/_impl/criteria.py
index 2f29f3e8..2178b35c 100644
--- a/pubtools/pulplib/_impl/criteria.py
+++ b/pubtools/pulplib/_impl/criteria.py
@@ -219,6 +219,28 @@ def in_(cls, values):
         """
         return InMatcher(values)
 
+    @classmethod
+    def less_than(cls, value):
+        """
+        Returns a matcher for a field whose value is less than the specified input
+        value.
+
+        Arguments:
+            value (object)
+                An object to match against the field
+
+        Example:
+            .. code-block:: python
+
+                # would match where last_publish is before "2019-08-27T00:00:00Z"
+                # date comparison requires a datetime.datetime object
+                crit = Criteria.with_field(
+                    'last_publish',
+                    Matcher.less_than(datetime.datetime(2019, 8, 27, 0, 0, 0))
+                )
+        """
+        return LessThanMatcher(value)
+
     def _map(self, _fn):
         # Internal-only: return self with matched value mapped through
         # the given function. Intended to be overridden in subclasses
@@ -265,6 +287,14 @@ class ExistsMatcher(Matcher):
     pass
 
 
+@attr.s
+class LessThanMatcher(Matcher):
+    _value = attr.ib()
+
+    def _map(self, fn):
+        return attr.evolve(self, value=fn(self._value))
+
+
 def coerce_to_matcher(value):
     if isinstance(value, Matcher):
         return value
diff --git a/pubtools/pulplib/_impl/fake/client.py b/pubtools/pulplib/_impl/fake/client.py
index aeb9ac36..97719d84 100644
--- a/pubtools/pulplib/_impl/fake/client.py
+++ b/pubtools/pulplib/_impl/fake/client.py
@@ -11,14 +11,14 @@
 
 from more_executors.futures import f_return, f_return_error, f_flat_map
 
-
 from pubtools.pulplib import (
     Page,
     PulpException,
     Criteria,
     Task,
-    MaintenanceReport,
     Repository,
+    Distributor,
+    MaintenanceReport,
 )
 from pubtools.pulplib._impl.client.search import filters_for_criteria
 from .. import compat_attr as attr
@@ -87,14 +87,34 @@ def search_repository(self, criteria=None):
         # callers should not make any assumption about the order of returned
         # values. Encourage that by returning output in unpredictable order
         random.shuffle(repos)
+        return self._prepare_pages(repos)
+
+    def search_distributor(self, criteria=None):
+        criteria = criteria or Criteria.true()
+        distributors = []
+
+        filters_for_criteria(criteria, Distributor)
+
+        try:
+            for repo in self._repositories:
+                for distributor in repo.distributors:
+                    if match_object(criteria, distributor):
+                        distributors.append(attr.evolve(distributor, repo_id=repo.id))
+        except Exception as ex:  # pylint: disable=broad-except
+            return f_return_error(ex)
+
+        random.shuffle(distributors)
+        return self._prepare_pages(distributors)
 
-        # Split it into pages
+    def _prepare_pages(self, resource_list):
+        # Split resource_list into pages
+        # resource_list: list of objects that paginated
         page_data = []
         current_page_data = []
-        while repos:
-            next_elem = repos.pop()
+        while resource_list:
+            next_elem = resource_list.pop()
             current_page_data.append(next_elem)
-            if len(current_page_data) == self._PAGE_SIZE and repos:
+            if len(current_page_data) == self._PAGE_SIZE and resource_list:
                 page_data.append(current_page_data)
                 current_page_data = []
 
diff --git a/pubtools/pulplib/_impl/fake/match.py b/pubtools/pulplib/_impl/fake/match.py
index 81b8bea1..5b4a305f 100644
--- a/pubtools/pulplib/_impl/fake/match.py
+++ b/pubtools/pulplib/_impl/fake/match.py
@@ -13,6 +13,7 @@
     InMatcher,
     RegexMatcher,
     ExistsMatcher,
+    LessThanMatcher,
 )
 from pubtools.pulplib._impl.model.common import PULP2_FIELD
 
@@ -45,11 +46,9 @@ def get_field(field, obj):
     # - If it's a field on the model, no conversion is needed since we already
     #   are storing plain objects from the model
     # - If it's a Pulp field, conversion will be handled in pulp_value
-    mapped_field, _ = map_field_for_type(field, matcher=None, type_hint=obj.__class__)
+    using_model_field = map_field_for_type(field, matcher=None, type_hint=obj.__class__)
 
     # Are we looking for a field on our model, or a raw Pulp field?
-    using_model_field = mapped_field is not field
-
     if using_model_field:
         # If matching a field on the model, we can simply grab and compare
         # the attribute directly.
@@ -124,6 +123,12 @@ def match_in(matcher, field, obj):
     return False
 
 
+@visit(LessThanMatcher)
+def match_field_less(matcher, field, obj):
+    value = get_field(field, obj)
+    return value < matcher._value
+
+
 def pulp_value(pulp_field, obj):
     # Given a Pulp 'field' name and a PulpObject instance,
     # returns the value on the object corresponding to that Pulp field.
diff --git a/pubtools/pulplib/_impl/model/distributor.py b/pubtools/pulplib/_impl/model/distributor.py
index 804f82c0..1fcf6744 100644
--- a/pubtools/pulplib/_impl/model/distributor.py
+++ b/pubtools/pulplib/_impl/model/distributor.py
@@ -26,6 +26,9 @@ class Distributor(PulpObject):
     of type `yum_distributor` may be used to create yum repositories.
     """
 
+    repo_id = pulp_attrib(type=str, default=None, pulp_field="repo_id")
+    """The :class:`pubtools.pulplib.Repository` ID this distributor is attached to."""
+
     last_publish = pulp_attrib(
         default=None, type=datetime.datetime, pulp_field="last_publish"
     )
diff --git a/pubtools/pulplib/_impl/model/repository/base.py b/pubtools/pulplib/_impl/model/repository/base.py
index f9373c7d..c1b99067 100644
--- a/pubtools/pulplib/_impl/model/repository/base.py
+++ b/pubtools/pulplib/_impl/model/repository/base.py
@@ -159,6 +159,20 @@ class Repository(PulpObject):
     _client = attr.ib(default=None, init=False, repr=False, cmp=False, hash=False)
     # hidden attribute for client attached to this object
 
+    @distributors.validator
+    def _check_repo_id(self, _, value):
+        # checks if distributor's repository id is same as the repository it
+        # is attached to
+        for distributor in value:
+            if not distributor.repo_id:
+                return
+            elif distributor.repo_id == self.id:
+                return
+            raise ValueError(
+                "repo_id doesn't match for %s. repo_id: %s, distributor.repo_id: %s"
+                % (distributor.id, self.id, distributor.repo_id)
+            )
+
     @property
     def _distributors_by_id(self):
         out = {}
diff --git a/pubtools/pulplib/_impl/schema/repository.yaml b/pubtools/pulplib/_impl/schema/repository.yaml
index 986d0e4b..affaecbe 100644
--- a/pubtools/pulplib/_impl/schema/repository.yaml
+++ b/pubtools/pulplib/_impl/schema/repository.yaml
@@ -17,6 +17,9 @@ definitions:
       distributor_type_id:
         # String ID of distributor's type, e.g. "yum_distributor"
         type: string
+      repo_id:
+        # String ID of distributor's repository
+        type: string
       config:
         # Config dict for this distributor, different per distributor type.
         # We won't mandate which config keys are used with each distributor,
diff --git a/setup.py b/setup.py
index 32f45961..579c9d97 100644
--- a/setup.py
+++ b/setup.py
@@ -21,7 +21,7 @@ def get_requirements():
 
 setup(
     name="pubtools-pulplib",
-    version="2.0.0",
+    version="2.1.0",
     packages=find_packages(exclude=["tests"]),
     package_data={"pubtools.pulplib._impl.schema": ["*.yaml"]},
     url="https://github.com/release-engineering/pubtools-pulplib",
diff --git a/tests/client/test_client.py b/tests/client/test_client.py
index f6106f5e..ae31ddb4 100644
--- a/tests/client/test_client.py
+++ b/tests/client/test_client.py
@@ -3,6 +3,7 @@
 import json
 import pytest
 import requests_mock
+
 from mock import patch
 
 from more_executors.futures import f_return
@@ -14,6 +15,7 @@
     PulpException,
     MaintenanceReport,
     Task,
+    Distributor,
 )
 
 
@@ -57,6 +59,33 @@ def test_can_search(client, requests_mocker):
     assert requests_mocker.call_count == 1
 
 
+def test_can_search_distributor(client, requests_mocker):
+    """search_distributor issues distributors/search POST request as expected."""
+    requests_mocker.post(
+        "https://pulp.example.com/pulp/api/v2/distributors/search/",
+        json=[
+            {
+                "id": "yum_distributor",
+                "distributor_type_id": "yum_distributor",
+                "repo_id": "test_rpm",
+            },
+            {"id": "cdn_distributor", "distributor_type_id": "rpm_rsync_distributor"},
+        ],
+    )
+
+    distributors_f = client.search_distributor()
+    distributors = [dist for dist in distributors_f.result().as_iter()]
+    # distributor objects are returned
+    assert sorted(distributors) == [
+        Distributor(id="cdn_distributor", type_id="rpm_rsync_distributor"),
+        Distributor(
+            id="yum_distributor", type_id="yum_distributor", repo_id="test_rpm"
+        ),
+    ]
+    # api is called once
+    assert requests_mocker.call_count == 1
+
+
 def test_search_retries(client, requests_mocker, caplog):
     """search_repository retries operations on failure."""
     logging.getLogger().setLevel(logging.WARNING)
diff --git a/tests/client/test_search.py b/tests/client/test_search.py
index 3c5cc87a..b3cfaeb6 100644
--- a/tests/client/test_search.py
+++ b/tests/client/test_search.py
@@ -1,4 +1,5 @@
 import pytest
+import datetime
 
 from pubtools.pulplib import Criteria, Matcher
 
@@ -58,6 +59,20 @@ def test_field_regex_criteria():
     ) == {"some.field": {"$regex": "abc"}}
 
 
+def test_field_less_than_criteria():
+    """with_field with less_than is translated as expected for
+    date and non-date types
+    """
+    publish_date = datetime.datetime(2019, 8, 27, 0, 0, 0)
+    c1 = Criteria.with_field("num_field", Matcher.less_than(5))
+    c2 = Criteria.with_field("date_field", Matcher.less_than(publish_date))
+
+    assert filters_for_criteria(c1) == {"num_field": {"$lt": 5}}
+    assert filters_for_criteria(c2) == {
+        "date_field": {"$lt": {"$date": "2019-08-27T00:00:00Z"}}
+    }
+
+
 def test_non_matcher():
     """field_match raises if invoked with something other than a matcher.
 
@@ -69,3 +84,16 @@ def test_non_matcher():
         field_match("oops not a matcher")
 
     assert "Not a matcher" in str(exc_info.value)
+
+
+def test_dict_matcher_value():
+    """criteria using a dict as matcher value"""
+
+    crit = Criteria.with_field(
+        "created",
+        Matcher.less_than({"created_date": datetime.datetime(2019, 9, 4, 0, 0, 0)}),
+    )
+
+    assert filters_for_criteria(crit) == {
+        "created": {"$lt": {"created_date": {"$date": "2019-09-04T00:00:00Z"}}}
+    }
diff --git a/tests/fake/test_fake_search.py b/tests/fake/test_fake_search.py
index 16a6687e..1eef8b01 100644
--- a/tests/fake/test_fake_search.py
+++ b/tests/fake/test_fake_search.py
@@ -2,7 +2,7 @@
 
 import pytest
 
-from pubtools.pulplib import FakeController, Repository, Criteria, Matcher
+from pubtools.pulplib import FakeController, Repository, Criteria, Matcher, Distributor
 
 
 def test_can_search_id():
@@ -158,7 +158,10 @@ def test_search_null_and():
     """Search with an empty AND gives an error."""
     controller = FakeController()
 
-    repo1 = Repository(id="repo1")
+    dist1 = Distributor(
+        id="yum_distributor", type_id="yum_distributor", repo_id="repo1"
+    )
+    repo1 = Repository(id="repo1", distributors=[dist1])
 
     controller.insert_repository(repo1)
 
@@ -167,6 +170,9 @@ def test_search_null_and():
     assert "Invalid AND in search query" in str(
         client.search_repository(crit).exception()
     )
+    assert "Invalid AND in search query" in str(
+        client.search_distributor(crit).exception()
+    )
 
 
 def test_search_null_or():
@@ -194,10 +200,14 @@ def test_search_bad_criteria():
 
     client = controller.client
 
-    with pytest.raises(Exception) as exc:
+    with pytest.raises(Exception) as repo_exc:
         client.search_repository("not a valid criteria")
 
-    assert "Not a criteria" in str(exc.value)
+    with pytest.raises(Exception) as dist_exc:
+        client.search_distributor("invalid criteria")
+
+    assert "Not a criteria" in str(repo_exc.value)
+    assert "Not a criteria" in str(dist_exc.value)
 
 
 def test_search_created_timestamp():
@@ -334,3 +344,52 @@ def test_search_paginates():
 
     # All repos should have been found
     assert sorted(found_repos) == sorted(repos)
+
+
+def test_search_distributor():
+    controller = FakeController()
+
+    dist1 = Distributor(
+        id="yum_distributor", type_id="yum_distributor", repo_id="repo1"
+    )
+    dist2 = Distributor(
+        id="cdn_distributor", type_id="rpm_rsync_distributor", repo_id="repo1"
+    )
+    repo1 = Repository(id="repo1", distributors=(dist1, dist2))
+
+    controller.insert_repository(repo1)
+
+    client = controller.client
+    crit = Criteria.true()
+
+    found = client.search_distributor(crit).result().data
+
+    assert sorted(found) == [dist2, dist1]
+
+
+def test_search_mapped_field_less_than():
+    controller = FakeController()
+
+    dist1 = Distributor(
+        id="yum_distributor",
+        type_id="yum_distributor",
+        repo_id="repo1",
+        last_publish=datetime.datetime(2019, 8, 23, 2, 5, 0, tzinfo=None),
+    )
+    dist2 = Distributor(
+        id="cdn_distributor",
+        type_id="rpm_rsync_distributor",
+        repo_id="repo1",
+        last_publish=datetime.datetime(2019, 8, 27, 2, 5, 0, tzinfo=None),
+    )
+    repo1 = Repository(id="repo1", distributors=(dist1, dist2))
+
+    controller.insert_repository(repo1)
+
+    client = controller.client
+    crit = Criteria.with_field(
+        "last_publish", Matcher.less_than(datetime.datetime(2019, 8, 24, 0, 0, 0))
+    )
+    found = client.search_distributor(crit).result().data
+
+    assert found == [dist1]
diff --git a/tests/repository/test_repository_from_data.py b/tests/repository/test_repository_from_data.py
index cd33f242..71600bb8 100644
--- a/tests/repository/test_repository_from_data.py
+++ b/tests/repository/test_repository_from_data.py
@@ -106,3 +106,13 @@ def test_distributors_last_publish_null():
     )
 
     assert repo.distributor("dist1").last_publish is None
+
+
+def test_invalid_distributor_repo_id():
+    """distributor's repo id being different from repo it's attached is invalid"""
+
+    dist = Distributor(id="dist", type_id="yum_distributor", repo_id="repo")
+    with pytest.raises(ValueError) as ex:
+        repo = Repository(id="test_repo", distributors=[dist])
+
+    assert "repo_id doesn't match for dist" in str(ex.value)