importd: when listing transfers, show progress percentage

With this change the pull protocol implementation processes will pass
progress data to importd which then passes this information on via the
bus. We use sd_notify() as generic transport for this communication,
making importd listen to them, while matching the incoming messages to
the right transfer.

10 files changed, 417 insertions, 15 deletions

diff --git a/Makefile.am b/Makefile.am
index 2b7ee6f478..45d7a349fd 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -5284,6 +5284,7 @@ systemd_importd_CFLAGS = \
 
 systemd_importd_LDADD = \
 	libsystemd-internal.la \
+	libsystemd-label.la \
 	libsystemd-shared.la
 
 systemd_pull_SOURCES = \
diff --git a/src/core/manager.c b/src/core/manager.c
index 4f771396a0..e2df91196e 100644
--- a/src/core/manager.c
+++ b/src/core/manager.c
@@ -84,9 +84,6 @@
 #define JOBS_IN_PROGRESS_PERIOD_USEC (USEC_PER_SEC / 3)
 #define JOBS_IN_PROGRESS_PERIOD_DIVISOR 3
 
-#define NOTIFY_FD_MAX 768
-#define NOTIFY_BUFFER_MAX PIPE_BUF
-
 static int manager_dispatch_notify_fd(sd_event_source *source, int fd, uint32_t revents, void *userdata);
 static int manager_dispatch_signal_fd(sd_event_source *source, int fd, uint32_t revents, void *userdata);
 static int manager_dispatch_time_change_fd(sd_event_source *source, int fd, uint32_t revents, void *userdata);
diff --git a/src/import/import-dkr.c b/src/import/import-dkr.c
index 24ba766b13..78ea80846c 100644
--- a/src/import/import-dkr.c
+++ b/src/import/import-dkr.c
@@ -22,7 +22,7 @@
 #include <curl/curl.h>
 #include <sys/prctl.h>
 
-#include "set.h"
+#include "sd-daemon.h"
 #include "json.h"
 #include "strv.h"
 #include "btrfs-util.h"
@@ -35,6 +35,14 @@
 #include "import-common.h"
 #include "import-dkr.h"
 
+typedef enum DkrProgress {
+        DKR_SEARCHING,
+        DKR_RESOLVING,
+        DKR_METADATA,
+        DKR_DOWNLOADING,
+        DKR_COPYING,
+} DkrProgress;
+
 struct DkrImport {
         sd_event *event;
         CurlGlue *glue;
@@ -56,6 +64,7 @@ struct DkrImport {
         char **response_registries;
 
         char **ancestry;
+        unsigned n_ancestry;
         unsigned current_ancestry;
 
         DkrImportFinished on_finished;
@@ -176,6 +185,53 @@ int dkr_import_new(
         return 0;
 }
 
+static void dkr_import_report_progress(DkrImport *i, DkrProgress p) {
+        unsigned percent;
+
+        assert(i);
+
+        switch (p) {
+
+        case DKR_SEARCHING:
+                percent = 0;
+                if (i->images_job)
+                        percent += i->images_job->progress_percent * 5 / 100;
+                break;
+
+        case DKR_RESOLVING:
+                percent = 5;
+                if (i->tags_job)
+                        percent += i->tags_job->progress_percent * 5 / 100;
+                break;
+
+        case DKR_METADATA:
+                percent = 10;
+                if (i->ancestry_job)
+                        percent += i->ancestry_job->progress_percent * 5 / 100;
+                if (i->json_job)
+                        percent += i->json_job->progress_percent * 5 / 100;
+                break;
+
+        case DKR_DOWNLOADING:
+                percent = 20;
+                percent += 75 * i->current_ancestry / MAX(1U, i->n_ancestry);
+                if (i->layer_job)
+                        percent += i->layer_job->progress_percent * 75 / MAX(1U, i->n_ancestry) / 100;
+
+                break;
+
+        case DKR_COPYING:
+                percent = 95;
+                break;
+
+        default:
+                assert_not_reached("Unknown progress state");
+        }
+
+        sd_notifyf(false, "X_IMPORT_PROGRESS=%u", percent);
+        log_debug("Combined progress %u%%", percent);
+}
+
 static int parse_id(const void *payload, size_t size, char **ret) {
         _cleanup_free_ char *buf = NULL, *id = NULL, *other = NULL;
         union json_value v = {};
@@ -438,6 +494,22 @@ static int dkr_import_job_on_open_disk(ImportJob *j) {
         return 0;
 }
 
+static void dkr_import_job_on_progress(ImportJob *j) {
+        DkrImport *i;
+
+        assert(j);
+        assert(j->userdata);
+
+        i = j->userdata;
+
+        dkr_import_report_progress(
+                        i,
+                        j == i->images_job                       ? DKR_SEARCHING :
+                        j == i->tags_job                         ? DKR_RESOLVING :
+                        j == i->ancestry_job || j == i->json_job ? DKR_METADATA :
+                                                                   DKR_DOWNLOADING);
+}
+
 static int dkr_import_pull_layer(DkrImport *i) {
         _cleanup_free_ char *path = NULL;
         const char *url, *layer = NULL;
@@ -488,6 +560,7 @@ static int dkr_import_pull_layer(DkrImport *i) {
 
         i->layer_job->on_finished = dkr_import_job_on_finished;
         i->layer_job->on_open_disk = dkr_import_job_on_open_disk;
+        i->layer_job->on_progress = dkr_import_job_on_progress;
 
         r = import_job_begin(i->layer_job);
         if (r < 0)
@@ -535,6 +608,7 @@ static void dkr_import_job_on_finished(ImportJob *j) {
                 }
 
                 log_info("Index lookup succeeded, directed to registry %s.", i->response_registries[0]);
+                dkr_import_report_progress(i, DKR_RESOLVING);
 
                 url = strappenda(PROTOCOL_PREFIX, i->response_registries[0], "/v1/repositories/", i->name, "/tags/", i->tag);
                 r = import_job_new(&i->tags_job, url, i->glue, i);
@@ -550,6 +624,7 @@ static void dkr_import_job_on_finished(ImportJob *j) {
                 }
 
                 i->tags_job->on_finished = dkr_import_job_on_finished;
+                i->tags_job->on_progress = dkr_import_job_on_progress;
 
                 r = import_job_begin(i->tags_job);
                 if (r < 0) {
@@ -575,6 +650,7 @@ static void dkr_import_job_on_finished(ImportJob *j) {
                 i->id = id;
 
                 log_info("Tag lookup succeeded, resolved to layer %s.", i->id);
+                dkr_import_report_progress(i, DKR_METADATA);
 
                 url = strappenda(PROTOCOL_PREFIX, i->response_registries[0], "/v1/images/", i->id, "/ancestry");
                 r = import_job_new(&i->ancestry_job, url, i->glue, i);
@@ -590,6 +666,7 @@ static void dkr_import_job_on_finished(ImportJob *j) {
                 }
 
                 i->ancestry_job->on_finished = dkr_import_job_on_finished;
+                i->ancestry_job->on_progress = dkr_import_job_on_progress;
 
                 url = strappenda(PROTOCOL_PREFIX, i->response_registries[0], "/v1/images/", i->id, "/json");
                 r = import_job_new(&i->json_job, url, i->glue, i);
@@ -605,6 +682,7 @@ static void dkr_import_job_on_finished(ImportJob *j) {
                 }
 
                 i->json_job->on_finished = dkr_import_job_on_finished;
+                i->json_job->on_progress = dkr_import_job_on_progress;
 
                 r = import_job_begin(i->ancestry_job);
                 if (r < 0) {
@@ -644,8 +722,11 @@ static void dkr_import_job_on_finished(ImportJob *j) {
 
                 strv_free(i->ancestry);
                 i->ancestry = ancestry;
-
+                i->n_ancestry = n;
                 i->current_ancestry = 0;
+
+                dkr_import_report_progress(i, DKR_DOWNLOADING);
+
                 r = dkr_import_pull_layer(i);
                 if (r < 0)
                         goto finish;
@@ -699,6 +780,8 @@ static void dkr_import_job_on_finished(ImportJob *j) {
         if (!dkr_import_is_done(i))
                 return;
 
+        dkr_import_report_progress(i, DKR_COPYING);
+
         r = dkr_import_make_local_copy(i);
         if (r < 0)
                 goto finish;
@@ -802,6 +885,7 @@ int dkr_import_pull(DkrImport *i, const char *name, const char *tag, const char
 
         i->images_job->on_finished = dkr_import_job_on_finished;
         i->images_job->on_header = dkr_import_job_on_header;
+        i->images_job->on_progress = dkr_import_job_on_progress;
 
         return import_job_begin(i->images_job);
 }
diff --git a/src/import/import-job.c b/src/import/import-job.c
index cde40b0f97..809486500b 100644
--- a/src/import/import-job.c
+++ b/src/import/import-job.c
@@ -63,6 +63,7 @@ static void import_job_finish(ImportJob *j, int ret) {
 
         if (ret == 0) {
                 j->state = IMPORT_JOB_DONE;
+                j->progress_percent = 100;
                 log_info("Download of %s complete.", j->url);
         } else {
                 j->state = IMPORT_JOB_FAILED;
@@ -621,6 +622,9 @@ static int import_job_progress_callback(void *userdata, curl_off_t dltotal, curl
 
                 j->progress_percent = percent;
                 j->last_status_usec = n;
+
+                if (j->on_progress)
+                        j->on_progress(j);
         }
 
         return 0;
diff --git a/src/import/import-job.h b/src/import/import-job.h
index 57090092ce..dcf89cb28c 100644
--- a/src/import/import-job.h
+++ b/src/import/import-job.h
@@ -33,7 +33,8 @@ typedef struct ImportJob ImportJob;
 
 typedef void (*ImportJobFinished)(ImportJob *job);
 typedef int (*ImportJobOpenDisk)(ImportJob *job);
-typedef int (*ImportJobHeader)(ImportJob*job, const char *header, size_t sz);
+typedef int (*ImportJobHeader)(ImportJob *job, const char *header, size_t sz);
+typedef void (*ImportJobProgress)(ImportJob *job);
 
 typedef enum ImportJobState {
         IMPORT_JOB_INIT,
@@ -66,6 +67,7 @@ struct ImportJob {
         ImportJobFinished on_finished;
         ImportJobOpenDisk on_open_disk;
         ImportJobHeader on_header;
+        ImportJobProgress on_progress;
 
         CurlGlue *glue;
         CURL *curl;
diff --git a/src/import/import-raw.c b/src/import/import-raw.c
index 21e2488d55..5c88cdb007 100644
--- a/src/import/import-raw.c
+++ b/src/import/import-raw.c
@@ -23,6 +23,7 @@
 #include <linux/fs.h>
 #include <curl/curl.h>
 
+#include "sd-daemon.h"
 #include "utf8.h"
 #include "strv.h"
 #include "copy.h"
@@ -37,7 +38,13 @@
 #include "import-common.h"
 #include "import-raw.h"
 
-typedef struct RawImportFile RawImportFile;
+typedef enum RawProgress {
+        RAW_DOWNLOADING,
+        RAW_VERIFYING,
+        RAW_UNPACKING,
+        RAW_FINALIZING,
+        RAW_COPYING,
+} RawProgress;
 
 struct RawImport {
         sd_event *event;
@@ -129,6 +136,57 @@ int raw_import_new(
         return 0;
 }
 
+static void raw_import_report_progress(RawImport *i, RawProgress p) {
+        unsigned percent;
+
+        assert(i);
+
+        switch (p) {
+
+        case RAW_DOWNLOADING: {
+                unsigned remain = 80;
+
+                percent = 0;
+
+                if (i->checksum_job) {
+                        percent += i->checksum_job->progress_percent * 5 / 100;
+                        remain -= 5;
+                }
+
+                if (i->signature_job) {
+                        percent += i->signature_job->progress_percent * 5 / 100;
+                        remain -= 5;
+                }
+
+                if (i->raw_job)
+                        percent += i->raw_job->progress_percent * remain / 100;
+                break;
+        }
+
+        case RAW_VERIFYING:
+                percent = 80;
+                break;
+
+        case RAW_UNPACKING:
+                percent = 85;
+                break;
+
+        case RAW_FINALIZING:
+                percent = 90;
+                break;
+
+        case RAW_COPYING:
+                percent = 95;
+                break;
+
+        default:
+                assert_not_reached("Unknown progress state");
+        }
+
+        sd_notifyf(false, "X_IMPORT_PROGRESS=%u", percent);
+        log_debug("Combined progress %u%%", percent);
+}
+
 static int raw_import_maybe_convert_qcow2(RawImport *i) {
         _cleanup_close_ int converted_fd = -1;
         _cleanup_free_ char *t = NULL;
@@ -304,14 +362,20 @@ static void raw_import_job_on_finished(ImportJob *j) {
                 /* This is a new download, verify it, and move it into place */
                 assert(i->raw_job->disk_fd >= 0);
 
+                raw_import_report_progress(i, RAW_VERIFYING);
+
                 r = import_verify(i->raw_job, i->checksum_job, i->signature_job);
                 if (r < 0)
                         goto finish;
 
+                raw_import_report_progress(i, RAW_UNPACKING);
+
                 r = raw_import_maybe_convert_qcow2(i);
                 if (r < 0)
                         goto finish;
 
+                raw_import_report_progress(i, RAW_FINALIZING);
+
                 r = import_make_read_only_fd(i->raw_job->disk_fd);
                 if (r < 0)
                         goto finish;
@@ -326,6 +390,8 @@ static void raw_import_job_on_finished(ImportJob *j) {
                 i->temp_path = NULL;
         }
 
+        raw_import_report_progress(i, RAW_COPYING);
+
         r = raw_import_make_local_copy(i);
         if (r < 0)
                 goto finish;
@@ -372,6 +438,17 @@ static int raw_import_job_on_open_disk(ImportJob *j) {
         return 0;
 }
 
+static void raw_import_job_on_progress(ImportJob *j) {
+        RawImport *i;
+
+        assert(j);
+        assert(j->userdata);
+
+        i = j->userdata;
+
+        raw_import_report_progress(i, RAW_DOWNLOADING);
+}
+
 int raw_import_pull(RawImport *i, const char *url, const char *local, bool force_local, ImportVerify verify) {
         int r;
 
@@ -401,6 +478,7 @@ int raw_import_pull(RawImport *i, const char *url, const char *local, bool force
 
         i->raw_job->on_finished = raw_import_job_on_finished;
         i->raw_job->on_open_disk = raw_import_job_on_open_disk;
+        i->raw_job->on_progress = raw_import_job_on_progress;
         i->raw_job->calc_checksum = verify != IMPORT_VERIFY_NO;
 
         r = import_find_old_etags(url, i->image_root, DT_REG, ".raw-", ".raw", &i->raw_job->old_etags);
@@ -416,12 +494,16 @@ int raw_import_pull(RawImport *i, const char *url, const char *local, bool force
                 return r;
 
         if (i->checksum_job) {
+                i->checksum_job->on_progress = raw_import_job_on_progress;
+
                 r = import_job_begin(i->checksum_job);
                 if (r < 0)
                         return r;
         }
 
         if (i->signature_job) {
+                i->signature_job->on_progress = raw_import_job_on_progress;
+
                 r = import_job_begin(i->signature_job);
                 if (r < 0)
                         return r;
diff --git a/src/import/import-tar.c b/src/import/import-tar.c
index 80ae83971e..999aa8ab5e 100644
--- a/src/import/import-tar.c
+++ b/src/import/import-tar.c
@@ -22,6 +22,7 @@
 #include <sys/prctl.h>
 #include <curl/curl.h>
 
+#include "sd-daemon.h"
 #include "utf8.h"
 #include "strv.h"
 #include "copy.h"
@@ -35,6 +36,13 @@
 #include "import-common.h"
 #include "import-tar.h"
 
+typedef enum TarProgress {
+        TAR_DOWNLOADING,
+        TAR_VERIFYING,
+        TAR_FINALIZING,
+        TAR_COPYING,
+} TarProgress;
+
 struct TarImport {
         sd_event *event;
         CurlGlue *glue;
@@ -134,6 +142,53 @@ int tar_import_new(
         return 0;
 }
 
+static void tar_import_report_progress(TarImport *i, TarProgress p) {
+        unsigned percent;
+
+        assert(i);
+
+        switch (p) {
+
+        case TAR_DOWNLOADING: {
+                unsigned remain = 85;
+
+                percent = 0;
+
+                if (i->checksum_job) {
+                        percent += i->checksum_job->progress_percent * 5 / 100;
+                        remain -= 5;
+                }
+
+                if (i->signature_job) {
+                        percent += i->signature_job->progress_percent * 5 / 100;
+                        remain -= 5;
+                }
+
+                if (i->tar_job)
+                        percent += i->tar_job->progress_percent * remain / 100;
+                break;
+        }
+
+        case TAR_VERIFYING:
+                percent = 85;
+                break;
+
+        case TAR_FINALIZING:
+                percent = 90;
+                break;
+
+        case TAR_COPYING:
+                percent = 95;
+                break;
+
+        default:
+                assert_not_reached("Unknown progress state");
+        }
+
+        sd_notifyf(false, "X_IMPORT_PROGRESS=%u", percent);
+        log_debug("Combined progress %u%%", percent);
+}
+
 static int tar_import_make_local_copy(TarImport *i) {
         int r;
 
@@ -209,10 +264,14 @@ static void tar_import_job_on_finished(ImportJob *j) {
         if (!i->tar_job->etag_exists) {
                 /* This is a new download, verify it, and move it into place */
 
+                tar_import_report_progress(i, TAR_VERIFYING);
+
                 r = import_verify(i->tar_job, i->checksum_job, i->signature_job);
                 if (r < 0)
                         goto finish;
 
+                tar_import_report_progress(i, TAR_FINALIZING);
+
                 r = import_make_read_only(i->temp_path);
                 if (r < 0)
                         goto finish;
@@ -226,6 +285,8 @@ static void tar_import_job_on_finished(ImportJob *j) {
                 i->temp_path = NULL;
         }
 
+        tar_import_report_progress(i, TAR_COPYING);
+
         r = tar_import_make_local_copy(i);
         if (r < 0)
                 goto finish;
@@ -277,6 +338,17 @@ static int tar_import_job_on_open_disk(ImportJob *j) {
         return 0;
 }
 
+static void tar_import_job_on_progress(ImportJob *j) {
+        TarImport *i;
+
+        assert(j);
+        assert(j->userdata);
+
+        i = j->userdata;
+
+        tar_import_report_progress(i, TAR_DOWNLOADING);
+}
+
 int tar_import_pull(TarImport *i, const char *url, const char *local, bool force_local, ImportVerify verify) {
         int r;
 
@@ -303,6 +375,7 @@ int tar_import_pull(TarImport *i, const char *url, const char *local, bool force
 
         i->tar_job->on_finished = tar_import_job_on_finished;
         i->tar_job->on_open_disk = tar_import_job_on_open_disk;
+        i->tar_job->on_progress = tar_import_job_on_progress;
         i->tar_job->calc_checksum = verify != IMPORT_VERIFY_NO;
 
         r = import_find_old_etags(url, i->image_root, DT_DIR, ".tar-", NULL, &i->tar_job->old_etags);
@@ -318,12 +391,16 @@ int tar_import_pull(TarImport *i, const char *url, const char *local, bool force
                 return r;
 
         if (i->checksum_job) {
+                i->checksum_job->on_progress = tar_import_job_on_progress;
+
                 r = import_job_begin(i->checksum_job);
                 if (r < 0)
                         return r;
         }
 
         if (i->signature_job) {
+                i->signature_job->on_progress = tar_import_job_on_progress;
+
                 r = import_job_begin(i->signature_job);
                 if (r < 0)
                         return r;
diff --git a/src/import/importd.c b/src/import/importd.c
index 3e417b1749..47157857c8 100644
--- a/src/import/importd.c
+++ b/src/import/importd.c
@@ -27,6 +27,8 @@
 #include "bus-util.h"
 #include "bus-common-errors.h"
 #include "def.h"
+#include "socket-util.h"
+#include "mkdir.h"
 #include "import-util.h"
 
 typedef struct Transfer Transfer;
@@ -66,6 +68,7 @@ struct Transfer {
         sd_event_source *log_event_source;
 
         unsigned n_canceled;
+        unsigned progress_percent;
 };
 
 struct Manager {
@@ -76,6 +79,10 @@ struct Manager {
         Hashmap *transfers;
 
         Hashmap *polkit_registry;
+
+        int notify_fd;
+
+        sd_event_source *notify_event_source;
 };
 
 #define TRANSFERS_MAX 64
@@ -395,7 +402,8 @@ static int transfer_start(Transfer *t) {
                 fd_cloexec(STDOUT_FILENO, false);
                 fd_cloexec(STDERR_FILENO, false);
 
-                putenv((char*) "SYSTEMD_LOG_TARGET=console-prefixed");
+                setenv("SYSTEMD_LOG_TARGET", "console-prefixed", 1);
+                setenv("NOTIFY_SOCKET", "/run/systemd/import/notify", 1);
 
                 cmd[k++] = import_verify_to_string(t->verify);
                 if (t->force_local)
@@ -453,6 +461,9 @@ static Manager *manager_unref(Manager *m) {
         if (!m)
                 return NULL;
 
+        sd_event_source_unref(m->notify_event_source);
+        safe_close(m->notify_fd);
+
         while ((t = hashmap_first(m->transfers)))
                 transfer_unref(t);
 
@@ -470,8 +481,107 @@ static Manager *manager_unref(Manager *m) {
 
 DEFINE_TRIVIAL_CLEANUP_FUNC(Manager*, manager_unref);
 
+static int manager_on_notify(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
+
+        char buf[NOTIFY_BUFFER_MAX+1];
+        struct iovec iovec = {
+                .iov_base = buf,
+                .iov_len = sizeof(buf)-1,
+        };
+        union {
+                struct cmsghdr cmsghdr;
+                uint8_t buf[CMSG_SPACE(sizeof(struct ucred)) +
+                            CMSG_SPACE(sizeof(int) * NOTIFY_FD_MAX)];
+        } control = {};
+        struct msghdr msghdr = {
+                .msg_iov = &iovec,
+                .msg_iovlen = 1,
+                .msg_control = &control,
+                .msg_controllen = sizeof(control),
+        };
+        struct ucred *ucred = NULL;
+        Manager *m = userdata;
+        struct cmsghdr *cmsg;
+        unsigned percent;
+        char *p, *e;
+        Transfer *t;
+        Iterator i;
+        ssize_t n;
+        int r;
+
+        n = recvmsg(fd, &msghdr, MSG_DONTWAIT|MSG_CMSG_CLOEXEC);
+        if (n < 0) {
+                if (errno == EAGAIN || errno == EINTR)
+                        return 0;
+
+                return -errno;
+        }
+
+        for (cmsg = CMSG_FIRSTHDR(&msghdr); cmsg; cmsg = CMSG_NXTHDR(&msghdr, cmsg)) {
+                if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
+                        close_many((int*) CMSG_DATA(cmsg), (cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int));
+                        log_warning("Somebody sent us unexpected fds, ignoring.");
+                        return 0;
+                } else if (cmsg->cmsg_level == SOL_SOCKET &&
+                           cmsg->cmsg_type == SCM_CREDENTIALS &&
+                           cmsg->cmsg_len == CMSG_LEN(sizeof(struct ucred))) {
+
+                        ucred = (struct ucred*) CMSG_DATA(cmsg);
+                }
+        }
+
+        if (msghdr.msg_flags & MSG_TRUNC) {
+                log_warning("Got overly long notification datagram, ignoring.");
+                return 0;
+        }
+
+        if (!ucred || ucred->pid <= 0) {
+                log_warning("Got notification datagram lacking credential information, ignoring.");
+                return 0;
+        }
+
+        HASHMAP_FOREACH(t, m->transfers, i)
+                if (ucred->pid == t->pid)
+                        break;
+
+        if (!t) {
+                log_warning("Got notification datagram from unexpected peer, ignoring.");
+                return 0;
+        }
+
+        buf[n] = 0;
+
+        p = startswith(buf, "X_IMPORT_PROGRESS=");
+        if (!p) {
+                p = strstr(buf, "\nX_IMPORT_PROGRESS=");
+                if (!p)
+                        return 0;
+
+                p += 19;
+        }
+
+        e = strchrnul(p, '\n');
+        *e = 0;
+
+        r = safe_atou(p, &percent);
+        if (r < 0 || percent > 100) {
+                log_warning("Got invalid percent value, ignoring.");
+                return 0;
+        }
+
+        t->progress_percent = percent;
+
+        log_debug("Got percentage from client: %u%%", percent);
+        return 0;
+}
+
 static int manager_new(Manager **ret) {
         _cleanup_(manager_unrefp) Manager *m = NULL;
+        static const union sockaddr_union sa = {
+                .un.sun_family = AF_UNIX,
+                .un.sun_path = "/run/systemd/import/notify",
+        };
+        static const int one = 1;
         int r;
 
         assert(ret);
@@ -490,6 +600,23 @@ static int manager_new(Manager **ret) {
         if (r < 0)
                 return r;
 
+        m->notify_fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
+        if (m->notify_fd < 0)
+                return -errno;
+
+        (void) mkdir_parents_label(sa.un.sun_path, 0755);
+        (void) unlink(sa.un.sun_path);
+
+        if (bind(m->notify_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path)) < 0)
+                return -errno;
+
+        if (setsockopt(m->notify_fd, SOL_SOCKET, SO_PASSCRED, &one, sizeof(one)) < 0)
+                return -errno;
+
+        r = sd_event_add_io(m->event, &m->notify_event_source, m->notify_fd, EPOLLIN, manager_on_notify, m);
+        if (r < 0)
+                return r;
+
         *ret = m;
         m = NULL;
 
@@ -698,7 +825,7 @@ static int method_list_transfers(sd_bus *bus, sd_bus_message *msg, void *userdat
         if (r < 0)
                 return r;
 
-        r = sd_bus_message_open_container(reply, 'a', "(ussso)");
+        r = sd_bus_message_open_container(reply, 'a', "(usssdo)");
         if (r < 0)
                 return r;
 
@@ -706,11 +833,12 @@ static int method_list_transfers(sd_bus *bus, sd_bus_message *msg, void *userdat
 
                 r = sd_bus_message_append(
                                 reply,
-                                "(ussso)",
+                                "(usssdo)",
                                 t->id,
                                 transfer_type_to_string(t->type),
                                 t->remote,
                                 t->local,
+                                (double) t->progress_percent / 100.0,
                                 t->object_path);
                 if (r < 0)
                         return r;
@@ -789,6 +917,24 @@ static int method_cancel_transfer(sd_bus *bus, sd_bus_message *msg, void *userda
         return sd_bus_reply_method_return(msg, NULL);
 }
 
+static int property_get_progress(
+                sd_bus *bus,
+                const char *path,
+                const char *interface,
+                const char *property,
+                sd_bus_message *reply,
+                void *userdata,
+                sd_bus_error *error) {
+
+        Transfer *t = userdata;
+
+        assert(bus);
+        assert(reply);
+        assert(t);
+
+        return sd_bus_message_append(reply, "d", (double) t->progress_percent / 100.0);
+}
+
 static BUS_DEFINE_PROPERTY_GET_ENUM(property_get_type, transfer_type, TransferType);
 static BUS_DEFINE_PROPERTY_GET_ENUM(property_get_verify, import_verify, ImportVerify);
 
@@ -799,6 +945,7 @@ static const sd_bus_vtable transfer_vtable[] = {
         SD_BUS_PROPERTY("Remote", "s", NULL, offsetof(Transfer, remote), SD_BUS_VTABLE_PROPERTY_CONST),
         SD_BUS_PROPERTY("Type", "s", property_get_type, offsetof(Transfer, type), SD_BUS_VTABLE_PROPERTY_CONST),
         SD_BUS_PROPERTY("Verify", "s", property_get_verify, offsetof(Transfer, verify), SD_BUS_VTABLE_PROPERTY_CONST),
+        SD_BUS_PROPERTY("Progress", "d", property_get_progress, 0, 0),
         SD_BUS_METHOD("Cancel", NULL, NULL, method_cancel, SD_BUS_VTABLE_UNPRIVILEGED),
         SD_BUS_SIGNAL("LogMessage", "us", 0),
         SD_BUS_VTABLE_END,
@@ -809,7 +956,7 @@ static const sd_bus_vtable manager_vtable[] = {
         SD_BUS_METHOD("PullTar", "sssb", "uo", method_pull_tar_or_raw, SD_BUS_VTABLE_UNPRIVILEGED),
         SD_BUS_METHOD("PullRaw", "sssb", "uo", method_pull_tar_or_raw, SD_BUS_VTABLE_UNPRIVILEGED),
         SD_BUS_METHOD("PullDkr", "sssssb", "uo", method_pull_dkr, SD_BUS_VTABLE_UNPRIVILEGED),
-        SD_BUS_METHOD("ListTransfers", NULL, "a(ussso)", method_list_transfers, SD_BUS_VTABLE_UNPRIVILEGED),
+        SD_BUS_METHOD("ListTransfers", NULL, "a(usssdo)", method_list_transfers, SD_BUS_VTABLE_UNPRIVILEGED),
         SD_BUS_METHOD("CancelTransfer", "u", NULL, method_cancel_transfer, SD_BUS_VTABLE_UNPRIVILEGED),
         SD_BUS_SIGNAL("TransferNew", "uo", 0),
         SD_BUS_SIGNAL("TransferRemoved", "uos", 0),
diff --git a/src/machine/machinectl.c b/src/machine/machinectl.c
index 95acc9315e..14cff6eabb 100644
--- a/src/machine/machinectl.c
+++ b/src/machine/machinectl.c
@@ -2071,6 +2071,7 @@ typedef struct TransferInfo {
         const char *type;
         const char *remote;
         const char *local;
+        double progress;
 } TransferInfo;
 
 static int compare_transfer_info(const void *a, const void *b) {
@@ -2088,6 +2089,7 @@ static int list_transfers(int argc, char *argv[], void *userdata) {
         const char *type, *remote, *local, *object;
         sd_bus *bus = userdata;
         uint32_t id, max_id = 0;
+        double progress;
         int r;
 
         pager_open_if_enabled();
@@ -2106,11 +2108,11 @@ static int list_transfers(int argc, char *argv[], void *userdata) {
                 return r;
         }
 
-        r = sd_bus_message_enter_container(reply, 'a', "(ussso)");
+        r = sd_bus_message_enter_container(reply, 'a', "(usssdo)");
         if (r < 0)
                 return bus_log_parse_error(r);
 
-        while ((r = sd_bus_message_read(reply, "(ussso)", &id, &type, &remote, &local, &object)) > 0) {
+        while ((r = sd_bus_message_read(reply, "(usssdo)", &id, &type, &remote, &local, &progress, &object)) > 0) {
                 size_t l;
 
                 if (!GREEDY_REALLOC(transfers, n_allocated, n_transfers + 1))
@@ -2120,6 +2122,7 @@ static int list_transfers(int argc, char *argv[], void *userdata) {
                 transfers[n_transfers].type = type;
                 transfers[n_transfers].remote = remote;
                 transfers[n_transfers].local = local;
+                transfers[n_transfers].progress = progress;
 
                 l = strlen(type);
                 if (l > max_type)
@@ -2148,15 +2151,17 @@ static int list_transfers(int argc, char *argv[], void *userdata) {
         qsort_safe(transfers, n_transfers, sizeof(TransferInfo), compare_transfer_info);
 
         if (arg_legend)
-                printf("%-*s %-*s %-*s %-*s\n",
+                printf("%-*s %-*s %-*s %-*s %-*s\n",
                        (int) MAX(2U, DECIMAL_STR_WIDTH(max_id)), "ID",
+                       (int) 7, "PERCENT",
                        (int) max_type, "TYPE",
                        (int) max_local, "LOCAL",
                        (int) max_remote, "REMOTE");
 
         for (j = 0; j < n_transfers; j++)
-                printf("%*" PRIu32 " %-*s %-*s %-*s\n",
+                printf("%*" PRIu32 " %*u%% %-*s %-*s %-*s\n",
                        (int) MAX(2U, DECIMAL_STR_WIDTH(max_id)), transfers[j].id,
+                       (int) 6, (unsigned) (transfers[j].progress * 100),
                        (int) max_type, transfers[j].type,
                        (int) max_local, transfers[j].local,
                        (int) max_remote, transfers[j].remote);
diff --git a/src/shared/def.h b/src/shared/def.h
index 96c45a6b72..a3d9fcf388 100644
--- a/src/shared/def.h
+++ b/src/shared/def.h
@@ -81,3 +81,6 @@
 #ifndef TTY_GID
 #define TTY_GID 5
 #endif
+
+#define NOTIFY_FD_MAX 768
+#define NOTIFY_BUFFER_MAX PIPE_BUF