notification_client.cc

Go to the documentation of this file.

#ifndef __STDC_FORMAT_MACROS
// NOLINTNEXTLINE
#define __STDC_FORMAT_MACROS
#endif

#include "notification_client.h"

#include <inttypes.h>

#include <string>
#include <vector>

#include "crypto/signature.h"
#include "manifest.h"
#include "manifest_fetch.h"
#include "notify/messages.h"
#include "notify/subscriber_sse.h"
#include "notify/subscriber_supervisor.h"
#include "supervisor.h"
#include "util/logging.h"
#include "util/posix.h"

namespace {

class ActivitySubscriber : public notify::SubscriberSSE {
 public:
  ActivitySubscriber(const std::string& server_url, FuseRemounter* remounter,
                     download::DownloadManager* dl_mgr,
                     signature::SignatureManager* sig_mgr)
      : SubscriberSSE(server_url),
        remounter_(remounter),
        dl_mgr_(dl_mgr),
        sig_mgr_(sig_mgr) {}

  virtual ~ActivitySubscriber() {}

  virtual notify::Subscriber::Status Consume(const std::string& repo_name,
                                             const std::string& msg_text) {
    notify::msg::Activity msg;
    if (!msg.FromJSONString(msg_text)) {
      LogCvmfs(kLogCvmfs, kLogSyslogErr,
               "NotificationClient - could not decode message.");
      return notify::Subscriber::kError;
    }

    manifest::ManifestEnsemble ensemble;
    manifest::Failures res =
        manifest::Verify(reinterpret_cast<unsigned char*>(&(msg.manifest_[0])),
                         msg.manifest_.size(), "",
                         repo_name, 0, NULL, sig_mgr_, dl_mgr_, &ensemble);

    if (res != manifest::kFailOk) {
      LogCvmfs(kLogCvmfs, kLogSyslogErr,
               "NotificationClient - manifest has invalid signature.");
      return notify::Subscriber::kError;
    }

    const UniquePtr<manifest::Manifest> manifest(manifest::Manifest::LoadMem(
        reinterpret_cast<const unsigned char*>(msg.manifest_.data()),
        msg.manifest_.size()));

    if (!manifest.IsValid()) {
      LogCvmfs(kLogCvmfs, kLogSyslogErr,
               "NotificationClient - could not parse manifest.");
      return notify::Subscriber::kError;
    }

    uint64_t new_revision = manifest->revision();
    LogCvmfs(kLogCvmfs, kLogSyslog,
             "NotificationClient - repository %s is now at revision %" PRIu64
             ", root hash: %s", repo_name.c_str(), new_revision,
             manifest->catalog_hash().ToString().c_str());

    FuseRemounter::Status status = remounter_->CheckSynchronously();
    switch (status) {
      case FuseRemounter::kStatusFailGeneral:
        LogCvmfs(kLogCvmfs, kLogSyslog, "NotificationClient - remount failed");
        break;
      case FuseRemounter::kStatusFailNoSpace:
        LogCvmfs(kLogCvmfs, kLogSyslog,
                 "NotificationClient - remount failed (no space)");
        break;
      case FuseRemounter::kStatusUp2Date:
        LogCvmfs(kLogCvmfs, kLogSyslog,
                 "NotificationClient - catalog up to date");
        break;
      case FuseRemounter::kStatusMaintenance:
        LogCvmfs(kLogCvmfs, kLogSyslog,
                 "NotificationClient - in maintenance mode");
        break;
      default:
        LogCvmfs(kLogCvmfs, kLogSyslog, "NotificationClient - internal error");
    }
    return notify::Subscriber::kContinue;
  }

 private:
  FuseRemounter* remounter_;
  download::DownloadManager* dl_mgr_;
  signature::SignatureManager* sig_mgr_;
};

}  // namespace

NotificationClient::NotificationClient(const std::string& config,
                                       const std::string& repo_name,
                                       FuseRemounter* remounter,
                                       download::DownloadManager* dl_mgr,
                                       signature::SignatureManager* sig_mgr)
    : config_(config),
      repo_name_(repo_name),
      remounter_(remounter),
      dl_mgr_(dl_mgr),
      sig_mgr_(sig_mgr),
      subscriber_(),
      thread_(),
      spawned_(false) {}

NotificationClient::~NotificationClient() {
  if (subscriber_.IsValid()) {
    subscriber_->Unsubscribe();
  }
  if (spawned_) {
    pthread_join(thread_, NULL);
    spawned_ = false;
  }
}

void NotificationClient::Spawn() {
  if (!spawned_) {
    if (pthread_create(&thread_, NULL, NotificationClient::Run, this)) {
      LogCvmfs(kLogCvmfs, kLogSyslogErr,
               "NotificationClient - Could not start background thread");
    }
    spawned_ = true;
  }
}

void* NotificationClient::Run(void* data) {
  NotificationClient* cl = static_cast<NotificationClient*>(data);

  cl->subscriber_ = new ActivitySubscriber(cl->config_, cl->remounter_,
                                           cl->dl_mgr_, cl->sig_mgr_);

  LogCvmfs(
      kLogCvmfs, kLogSyslog,
      "NotificationClient - Entering subscription loop for repository: %s.",
      cl->repo_name_.c_str());

  // Retry settings: accept no more than 10 failures in the last minute
  const int num_retries = 10;
  const uint64_t interval = 60;
  notify::SubscriberSupervisor supervisor(
      cl->subscriber_.weak_ref(), cl->repo_name_, num_retries, interval);
  supervisor.Run();

  return NULL;
}