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.

diff --git a/Makefile.am b/Makefile.am
index 2b7ee6f4789541d41eb6f06f374c0040bbdb3539..45d7a349fdac85bf50397429a52372bec984e653 100644 (file)
--- 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 4f771396a0ceb5fb59cbc8e49415b58e3567e4eb..e2df91196e58f2f4020f3f12408ff2f612c0af7e 100644 (file)
--- 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 24ba766b1383c34f4ae940092abd8e49791115a8..78ea80846c7fa12698a66cbfb1719e2b42cd66a0 100644 (file)
--- 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 cde40b0f97c349679c6a44495359d448b0268f07..809486500ba55cf6d6b045e5b1fda78956560d3f 100644 (file)
--- 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 57090092ce35fb1710ad2ab74ba002697c5456f4..dcf89cb28c45314a92879b4df88ef546770e1405 100644 (file)
--- 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 21e2488d55a59417ff260cd6532bdebf3a8e61a3..5c88cdb007c836830a525a36855e8a9f942c4479 100644 (file)
--- 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 80ae83971efb93dd03164584bda76dff2c4d1797..999aa8ab5e8c5dd9924402e7fe36b64e9cc97a47 100644 (file)
--- 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 3e417b174964a863654050c3c61a7bc3c652b339..47157857c8293ac055c8660087f88d1b71602719 100644 (file)
--- 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 95acc9315e921f144b23df3c7f01dbb80bfbd025..14cff6eabb6d0d6bf3e8af63b43524b96c6f54e3 100644 (file)
--- 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 96c45a6b723dededf0c59e0b08778d124fd04779..a3d9fcf388c8783bc826bb2d5fa53db19944c053 100644 (file)
--- 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