remoting/host/heartbeat_sender.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "remoting/host/heartbeat_sender.h"
   6
   7 #include <math.h>
   8
   9 #include "base/bind.h"
  10 #include "base/callback_helpers.h"
  11 #include "base/rand_util.h"
  12 #include "base/strings/string_number_conversions.h"
  13 #include "base/strings/stringize_macros.h"
  14 #include "base/thread_task_runner_handle.h"
  15 #include "base/time/time.h"
  16 #include "remoting/base/constants.h"
  17 #include "remoting/base/logging.h"
  18 #include "remoting/host/server_log_entry_host.h"
  19 #include "remoting/signaling/iq_sender.h"
  20 #include "remoting/signaling/jid_util.h"
  21 #include "remoting/signaling/server_log_entry.h"
  22 #include "remoting/signaling/signal_strategy.h"
  23 #include "third_party/webrtc/libjingle/xmllite/xmlelement.h"
  24 #include "third_party/webrtc/libjingle/xmpp/constants.h"
  25
  26 using buzz::QName;
  27 using buzz::XmlElement;
  28
  29 namespace remoting {
  30
  31 namespace {
  32
  33 const char kHeartbeatQueryTag[] = "heartbeat";
  34 const char kHostIdAttr[] = "hostid";
  35 const char kHostVersionTag[] = "host-version";
  36 const char kHeartbeatSignatureTag[] = "signature";
  37 const char kSequenceIdAttr[] = "sequence-id";
  38 const char kHostOfflineReasonAttr[] = "host-offline-reason";
  39
  40 const char kErrorTag[] = "error";
  41 const char kNotFoundTag[] = "item-not-found";
  42
  43 const char kHeartbeatResultTag[] = "heartbeat-result";
  44 const char kSetIntervalTag[] = "set-interval";
  45 const char kExpectedSequenceIdTag[] = "expected-sequence-id";
  46
  47 const int64 kDefaultHeartbeatIntervalMs = 5 * 60 * 1000;  // 5 minutes.
  48 const int64 kResendDelayMs = 10 * 1000;  // 10 seconds.
  49 const int64 kResendDelayOnHostNotFoundMs = 10 * 1000; // 10 seconds.
  50 const int kMaxResendOnHostNotFoundCount = 12;  // 2 minutes (12 x 10 seconds).
  51
  52 }  // namespace
  53
  54 HeartbeatSender::HeartbeatSender(
  55     const base::Closure& on_heartbeat_successful_callback,
  56     const base::Closure& on_unknown_host_id_error,
  57     const std::string& host_id,
  58     SignalStrategy* signal_strategy,
  59     const scoped_refptr<const RsaKeyPair>& host_key_pair,
  60     const std::string& directory_bot_jid)
  61     : on_heartbeat_successful_callback_(on_heartbeat_successful_callback),
  62       on_unknown_host_id_error_(on_unknown_host_id_error),
  63       host_id_(host_id),
  64       signal_strategy_(signal_strategy),
  65       host_key_pair_(host_key_pair),
  66       directory_bot_jid_(directory_bot_jid),
  67       interval_ms_(kDefaultHeartbeatIntervalMs),
  68       sequence_id_(0),
  69       sequence_id_was_set_(false),
  70       sequence_id_recent_set_num_(0),
  71       heartbeat_succeeded_(false),
  72       failed_startup_heartbeat_count_(0) {
  73   DCHECK(signal_strategy_);
  74   DCHECK(host_key_pair_.get());
  75   DCHECK(thread_checker_.CalledOnValidThread());
  76
  77   signal_strategy_->AddListener(this);
  78
  79   // Start heartbeats if the |signal_strategy_| is already connected.
  80   OnSignalStrategyStateChange(signal_strategy_->GetState());
  81 }
  82
  83 HeartbeatSender::~HeartbeatSender() {
  84   signal_strategy_->RemoveListener(this);
  85 }
  86
  87 void HeartbeatSender::OnSignalStrategyStateChange(SignalStrategy::State state) {
  88   DCHECK(thread_checker_.CalledOnValidThread());
  89   if (state == SignalStrategy::CONNECTED) {
  90     iq_sender_.reset(new IqSender(signal_strategy_));
  91     SendStanza();
  92     timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(interval_ms_),
  93                  this, &HeartbeatSender::SendStanza);
  94   } else if (state == SignalStrategy::DISCONNECTED) {
  95     request_.reset();
  96     iq_sender_.reset();
  97     timer_.Stop();
  98     timer_resend_.Stop();
  99   }
 100 }
 101
 102 bool HeartbeatSender::OnSignalStrategyIncomingStanza(
 103     const buzz::XmlElement* stanza) {
 104   return false;
 105 }
 106
 107 void HeartbeatSender::OnHostOfflineReasonTimeout() {
 108   DCHECK(!host_offline_reason_ack_callback_.is_null());
 109
 110   base::ResetAndReturn(&host_offline_reason_ack_callback_).Run(false);
 111 }
 112
 113 void HeartbeatSender::OnHostOfflineReasonAck() {
 114   if (host_offline_reason_ack_callback_.is_null()) {
 115     DCHECK(!host_offline_reason_timeout_timer_.IsRunning());
 116     return;
 117   }
 118
 119   DCHECK(host_offline_reason_timeout_timer_.IsRunning());
 120   host_offline_reason_timeout_timer_.Stop();
 121
 122   // Run the ACK callback under a clean stack via PostTask() (because the
 123   // callback can end up deleting |this| HeartbeatSender [i.e. when used from
 124   // HostSignalingManager]).
 125   base::ThreadTaskRunnerHandle::Get()->PostTask(
 126       FROM_HERE,
 127       base::Bind(base::ResetAndReturn(&host_offline_reason_ack_callback_),
 128                  true));
 129 }
 130
 131 void HeartbeatSender::SetHostOfflineReason(
 132     const std::string& host_offline_reason,
 133     const base::TimeDelta& timeout,
 134     const base::Callback<void(bool success)>& ack_callback) {
 135   DCHECK(thread_checker_.CalledOnValidThread());
 136   DCHECK(host_offline_reason_ack_callback_.is_null());
 137   host_offline_reason_ = host_offline_reason;
 138   host_offline_reason_ack_callback_ = ack_callback;
 139   host_offline_reason_timeout_timer_.Start(
 140       FROM_HERE, timeout, this, &HeartbeatSender::OnHostOfflineReasonTimeout);
 141   if (signal_strategy_->GetState() == SignalStrategy::CONNECTED) {
 142     DoSendStanza();
 143   }
 144 }
 145
 146 void HeartbeatSender::SendStanza() {
 147   // Make sure we don't send another heartbeat before the heartbeat interval
 148   // has expired.
 149   timer_resend_.Stop();
 150   DoSendStanza();
 151 }
 152
 153 void HeartbeatSender::ResendStanza() {
 154   // Make sure we don't send another heartbeat before the heartbeat interval
 155   // has expired.
 156   timer_.Reset();
 157   DoSendStanza();
 158 }
 159
 160 void HeartbeatSender::DoSendStanza() {
 161   DCHECK(thread_checker_.CalledOnValidThread());
 162   DCHECK(signal_strategy_->GetState() == SignalStrategy::CONNECTED);
 163   VLOG(1) << "Sending heartbeat stanza to " << directory_bot_jid_;
 164
 165   request_ = iq_sender_->SendIq(
 166       buzz::STR_SET, directory_bot_jid_, CreateHeartbeatMessage(),
 167       base::Bind(&HeartbeatSender::ProcessResponse,
 168                  base::Unretained(this),
 169                  !host_offline_reason_.empty()));
 170   ++sequence_id_;
 171 }
 172
 173 void HeartbeatSender::ProcessResponse(
 174     bool is_offline_heartbeat_response,
 175     IqRequest* request,
 176     const XmlElement* response) {
 177   DCHECK(thread_checker_.CalledOnValidThread());
 178
 179   std::string type = response->Attr(buzz::QN_TYPE);
 180   if (type == buzz::STR_ERROR) {
 181     const XmlElement* error_element =
 182         response->FirstNamed(QName(buzz::NS_CLIENT, kErrorTag));
 183     if (error_element) {
 184       if (error_element->FirstNamed(QName(buzz::NS_STANZA, kNotFoundTag))) {
 185         LOG(ERROR) << "Received error: Host ID not found";
 186         // If the host was registered immediately before it sends a heartbeat,
 187         // then server-side latency may prevent the server recognizing the
 188         // host ID in the heartbeat. So even if all of the first few heartbeats
 189         // get a "host ID not found" error, that's not a good enough reason to
 190         // exit.
 191         failed_startup_heartbeat_count_++;
 192         if (!heartbeat_succeeded_ && (failed_startup_heartbeat_count_ <=
 193                 kMaxResendOnHostNotFoundCount)) {
 194           timer_resend_.Start(FROM_HERE,
 195                               base::TimeDelta::FromMilliseconds(
 196                                   kResendDelayOnHostNotFoundMs),
 197                               this,
 198                               &HeartbeatSender::ResendStanza);
 199           return;
 200         }
 201         on_unknown_host_id_error_.Run();
 202         return;
 203       }
 204     }
 205
 206     LOG(ERROR) << "Received error in response to heartbeat: "
 207                << response->Str();
 208     return;
 209   }
 210
 211   // This method must only be called for error or result stanzas.
 212   DCHECK_EQ(std::string(buzz::STR_RESULT), type);
 213
 214   const XmlElement* result_element =
 215       response->FirstNamed(QName(kChromotingXmlNamespace, kHeartbeatResultTag));
 216   if (result_element) {
 217     const XmlElement* set_interval_element =
 218         result_element->FirstNamed(QName(kChromotingXmlNamespace,
 219                                          kSetIntervalTag));
 220     if (set_interval_element) {
 221       const std::string& interval_str = set_interval_element->BodyText();
 222       int interval;
 223       if (!base::StringToInt(interval_str, &interval) || interval <= 0) {
 224         LOG(ERROR) << "Received invalid set-interval: "
 225                    << set_interval_element->Str();
 226       } else {
 227         SetInterval(interval * base::Time::kMillisecondsPerSecond);
 228       }
 229     }
 230
 231     bool did_set_sequence_id = false;
 232     const XmlElement* expected_sequence_id_element =
 233         result_element->FirstNamed(QName(kChromotingXmlNamespace,
 234                                          kExpectedSequenceIdTag));
 235     if (expected_sequence_id_element) {
 236       // The sequence ID sent in the previous heartbeat was not what the server
 237       // expected, so send another heartbeat with the expected sequence ID.
 238       const std::string& expected_sequence_id_str =
 239           expected_sequence_id_element->BodyText();
 240       int expected_sequence_id;
 241       if (!base::StringToInt(expected_sequence_id_str, &expected_sequence_id)) {
 242         LOG(ERROR) << "Received invalid " << kExpectedSequenceIdTag << ": " <<
 243             expected_sequence_id_element->Str();
 244       } else {
 245         SetSequenceId(expected_sequence_id);
 246         sequence_id_recent_set_num_++;
 247         did_set_sequence_id = true;
 248       }
 249     }
 250     if (!did_set_sequence_id) {
 251       // It seems the bot accepted our signature and our message.
 252       sequence_id_recent_set_num_ = 0;
 253
 254       // Notify listener of the first successful heartbeat.
 255       if (!heartbeat_succeeded_) {
 256         on_heartbeat_successful_callback_.Run();
 257       }
 258       heartbeat_succeeded_ = true;
 259
 260       // Notify caller of SetHostOfflineReason that we got an ack.
 261       if (is_offline_heartbeat_response) {
 262         OnHostOfflineReasonAck();
 263       }
 264     }
 265   }
 266 }
 267
 268 void HeartbeatSender::SetInterval(int interval) {
 269   if (interval != interval_ms_) {
 270     interval_ms_ = interval;
 271
 272     // Restart the timer with the new interval.
 273     if (timer_.IsRunning()) {
 274       timer_.Stop();
 275       timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(interval_ms_),
 276                    this, &HeartbeatSender::SendStanza);
 277     }
 278   }
 279 }
 280
 281 void HeartbeatSender::SetSequenceId(int sequence_id) {
 282   sequence_id_ = sequence_id;
 283   // Setting the sequence ID may be a symptom of a temporary server-side
 284   // problem, which would affect many hosts, so don't send a new heartbeat
 285   // immediately, as many hosts doing so may overload the server.
 286   // But the server will usually set the sequence ID when it receives the first
 287   // heartbeat from a host. In that case, we can send a new heartbeat
 288   // immediately, as that only happens once per host instance.
 289   if (!sequence_id_was_set_) {
 290     ResendStanza();
 291   } else {
 292     HOST_LOG << "The heartbeat sequence ID has been set more than once: "
 293               << "the new value is " << sequence_id;
 294     double delay = pow(2.0, sequence_id_recent_set_num_) *
 295         (1 + base::RandDouble()) * kResendDelayMs;
 296     if (delay <= interval_ms_) {
 297       timer_resend_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(delay),
 298                           this, &HeartbeatSender::ResendStanza);
 299     }
 300   }
 301   sequence_id_was_set_ = true;
 302 }
 303
 304 scoped_ptr<XmlElement> HeartbeatSender::CreateHeartbeatMessage() {
 305   // Create heartbeat stanza.
 306   scoped_ptr<XmlElement> heartbeat(new XmlElement(
 307       QName(kChromotingXmlNamespace, kHeartbeatQueryTag)));
 308   heartbeat->AddAttr(QName(kChromotingXmlNamespace, kHostIdAttr), host_id_);
 309   heartbeat->AddAttr(QName(kChromotingXmlNamespace, kSequenceIdAttr),
 310                  base::IntToString(sequence_id_));
 311   if (!host_offline_reason_.empty()) {
 312     heartbeat->AddAttr(
 313         QName(kChromotingXmlNamespace, kHostOfflineReasonAttr),
 314         host_offline_reason_);
 315   }
 316   heartbeat->AddElement(CreateSignature().release());
 317   // Append host version.
 318   scoped_ptr<XmlElement> version_tag(new XmlElement(
 319       QName(kChromotingXmlNamespace, kHostVersionTag)));
 320   version_tag->AddText(STRINGIZE(VERSION));
 321   heartbeat->AddElement(version_tag.release());
 322   // Append log message (which isn't signed).
 323   scoped_ptr<XmlElement> log(ServerLogEntry::MakeStanza());
 324   scoped_ptr<ServerLogEntry> log_entry(MakeLogEntryForHeartbeat());
 325   AddHostFieldsToLogEntry(log_entry.get());
 326   log->AddElement(log_entry->ToStanza().release());
 327   heartbeat->AddElement(log.release());
 328   return heartbeat.Pass();
 329 }
 330
 331 scoped_ptr<XmlElement> HeartbeatSender::CreateSignature() {
 332   scoped_ptr<XmlElement> signature_tag(new XmlElement(
 333       QName(kChromotingXmlNamespace, kHeartbeatSignatureTag)));
 334
 335   std::string message = NormalizeJid(signal_strategy_->GetLocalJid()) + ' ' +
 336                         base::IntToString(sequence_id_);
 337   std::string signature(host_key_pair_->SignMessage(message));
 338   signature_tag->AddText(signature);
 339
 340   return signature_tag.Pass();
 341 }
 342
 343 }  // namespace remoting