Merge ubuntu-debuginfod:poller into ubuntu-debuginfod:master

[Sergio Durigan Junior]

This MP implements the poller for debuginfod. There are 2 classes implemented here: DebugPoller, which offers the basic functionality for a poller, and DdebPoller, which extends the former and does the actual polling for newly published ddebs on LP. The intention is that, in the future, we will have more classes for other types of debug artifacts that will need to be fetched.

[Bryce Harrington (bryce)]

I have several suggestions for improvements but none that I feel to be critical enough to block on so set as approved, and you can take or leave the suggestions as desired. The code looks good, this will add a nice functionality!

Don't forget to also include the python3-launchpad dependency in the README.md. (Might not hurt to have some discussion in README.md or somewhere about how these new classes are used, but maybe that comes later.)

review: Approve

[Sergio Durigan Junior (sergiodj)]

Thanks for the review, Bryce. Comments inline.

[Lena Voytek (lvoytek)]

Logic looks good to me! Added some comments on formatting and some nitpicks

review: Approve

[Sergio Durigan Junior (sergiodj)]

Thanks for the review, Lena.

diff --git a/ddebpoller.py b/ddebpoller.py
new file mode 100644
index 0000000..a19d92c
--- /dev/null
+++ b/ddebpoller.py
@@ -0,0 +1,75 @@
+#!/usr/bin/python3
+
+# Copyright (C) 2022 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+# Authors: Sergio Durigan Junior <[e-mail]>
+
+from debugpoller import DebugPoller
+
+
+class DdebPoller(DebugPoller):
+    def __init__(
+        self, initial_interval=None, force_initial_interval=False, dry_run=False
+    ):
+        """Initialize the object using 'ddeb' as its name.
+
+        Look at DebugPoller's docstring for an explanation about the arguments."""
+        super().__init__(
+            "ddeb",
+            initial_interval=initial_interval,
+            force_initial_interval=force_initial_interval,
+            dry_run=dry_run,
+        )
+
+    def get_ddebs(self):
+        """Get the list of ddebs that have been published since the last
+        timestamp from Launchpad."""
+        timestamp = self._get_timestamp()
+        assert timestamp is not None, f"_get_timestamp returned None"
+        assert timestamp != "", f"_get_timestamp returned a blank timestamp"
+        new_timestamp = self._generate_timestamp()
+        assert new_timestamp is not None, f"_generate_timestamp returned None"
+        assert new_timestamp != "", f"_generate_timestamp returned a blank timestamp"
+
+        result = []
+
+        self._logger.info(
+            f"Polling ddebs created since '{timestamp}' (it's now '{new_timestamp}')"
+        )
+
+        for pkg in self._main_archive.getPublishedSources(
+            order_by_date=True, created_since_date=timestamp, status="Published"
+        ):
+            ddeb_urls = [url for url in pkg.binaryFileUrls() if url.endswith(".ddeb")]
+            ddebs_len = len(ddeb_urls)
+            if ddebs_len > 0:
+                distro_series = self._lp.load(pkg.distro_series_link).name
+                pkgname = pkg.source_package_name
+                pkgver = pkg.source_package_version
+                msg = {
+                    "source_package": pkgname,
+                    "version": pkgver,
+                    "component": pkg.component_name,
+                    "distro_series": distro_series,
+                    "ddebs": ddeb_urls,
+                }
+                self._logger.debug(
+                    f"For source package '{pkgname}-{pkgver}', found {ddebs_len}:\n{ddeb_urls}"
+                )
+                result.append(msg)
+
+        self._record_timestamp(new_timestamp)
+        return result
diff --git a/debugpoller.py b/debugpoller.py
new file mode 100644
index 0000000..77cb2a1
--- /dev/null
+++ b/debugpoller.py
@@ -0,0 +1,135 @@
+#!/usr/bin/python3
+
+# Copyright (C) 2022 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+# Authors: Sergio Durigan Junior <[e-mail]>
+
+import os
+import sys
+from launchpadlib.launchpad import Launchpad
+import datetime
+import logging
+
+
+class DebugPoller:
+    """The DebugPoller class implements the basics of a poller for the
+    debuginfod service."""
+
+    # The timestamp file we will save our timestamp into.
+    TIMESTAMP_FILE = os.path.expanduser("~/.config/ubuntu-debuginfod/timestamp")
+
+    # The timestamp format we use.  Launchpad requires this format.
+    TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S"
+
+    def __init__(
+        self,
+        module_name,
+        initial_interval=1,
+        force_initial_interval=False,
+        dry_run=False,
+    ):
+        """Initalize the DebugPoller object.
+
+        :param str module_name: Name of the DebugPoller module.  This
+            is used to compose the timestamp filename so that it is
+            unique.
+
+        :param int initial_interval: The initial interval (in hours),
+            i.e., if no timestamp file has been found then we use this
+            value to tell Launchpad that we want files created since
+            (NOW - INITIAL_INTERVAL).  Default is 1 hour.
+
+        :param bool force_initial_interval: Force the poller to use
+            the initial interval to calculate the timestamp, instead
+            of relying on the timestamp file.  Default is False.
+
+        :param bool dry_run: Tell the poller that it shouldn't record
+            the timestamp in the file when the operation finishes.
+            Default is False.
+        """
+        self._lp = Launchpad.login_anonymously("ubuntu-debuginfod poller", "production")
+        self._main_archive = self._lp.distributions["ubuntu"].main_archive
+        self.TIMESTAMP_FORMAT = f"{self.TIMESTAMP_FORMAT}-{module_name}"
+
+        self._logger = logging.getLogger(__name__)
+        logging.basicConfig(
+            level=logging.INFO,
+            format=(
+                "%(levelname) -10s %(asctime)s %(name) -20s %(funcName) "
+                "-25s : %(message)s"
+            ),
+        )
+
+        self._initial_interval = initial_interval
+        self._force_initial_interval = force_initial_interval
+        self._dry_run = dry_run
+
+    def _generate_timestamp(self, interval=None):
+        """Generate a timestamp that can be used when querying Launchpad
+        (via getPublishedSources).
+
+        :param interval: Specify how long ago (in hours) the timestamp must
+            refer to.  If not specified, the timestamp is generated for
+            the current time.
+        :type interval: int or None
+        """
+        d = datetime.datetime.now(datetime.timezone.utc)
+
+        if interval is not None:
+            self._logger.debug(f"Generating timestamp with interval {interval}")
+            d = d - datetime.timedelta(hours=interval)
+
+        self._logger.debug(f"Generated timestamp '{d.strftime(self.TIMESTAMP_FORMAT)}'")
+        return d.strftime(self.TIMESTAMP_FORMAT)
+
+    def _get_timestamp(self):
+        """Get the timestamp from the timestamp file, or generate a
+        new one.
+
+        If a timestamp file exists, returns its value.  Otherwise,
+        generate a new one with self._initial_interval.
+        """
+        if not os.path.exists(self.TIMESTAMP_FILE) or self._force_initial_interval:
+            self._logger.debug(f"Timestamp file '{self.TIMESTAMP_FILE}' doesn't exist")
+            self._logger.debug(
+                f"Or force_initial_interval = {self._force_initial_interval}"
+            )
+            return self._generate_timestamp(interval=self._initial_interval)
+
+        with open(self.TIMESTAMP_FILE, "r", encoding="UTF-8") as f:
+            return f.readline()
+
+    def _record_timestamp(self, timestamp):
+        """Save the timestamp into the timestamp file.
+
+        :param str timestamp: Timestamp that should be saved into the
+            TIMESTAMP_FILE.
+        """
+        if self._dry_run:
+            self._logger.debug("dry_run enabled, not recording timestamp file")
+            return
+
+        dirname = os.path.dirname(self.TIMESTAMP_FILE)
+        os.makedirs(dirname, mode=0o755, exist_ok=True)
+        with open(self.TIMESTAMP_FILE, "w", encoding="UTF-8") as f:
+            f.write(timestamp)
+
+    def set_initial_interval(self, initial_interval):
+        """Set the initial_interval value.
+
+        :param int initial_interval: The new initial_interval to be used.
+        """
+        self._initial_interval = initial_interval