Initial RemoteCommandService

components/policy/core/common/remote_commands/remote_commands_service.cc

Index: components/policy/core/common/remote_commands/remote_commands_service.cc
diff --git a/components/policy/core/common/remote_commands/remote_commands_service.cc b/components/policy/core/common/remote_commands/remote_commands_service.cc
new file mode 100644
index 0000000000000000000000000000000000000000..7478026c60023cdb7e1bdbc7d004588430d44b33
--- /dev/null
+++ b/components/policy/core/common/remote_commands/remote_commands_service.cc
@@ -0,0 +1,154 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "components/policy/core/common/remote_commands/remote_commands_service.h"
+
+#include <algorithm>
+#include <string>
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/location.h"
+#include "base/logging.h"
+#include "base/time/time.h"
+#include "components/policy/core/common/cloud/cloud_policy_client.h"
+#include "components/policy/core/common/cloud/device_management_service.h"
+
+namespace {
+
+// XXX
+const int kMaxUnhandledFetchRequests = 30;
+
+// XXX
+const int kAutoFetchingIntervalInMinutes = 30;

[bartfab (slow), 2015/01/29 10:18:05]

The server should send us a Tango invalidation when there are new commands.
Auto-fetching would only be a backup that should run rarely, e.g. every 24
hours. Since commands expire after 3 hours, the auto-fetching would actually be
quite useless. I think we can remove it completely and fetch on invalidaiton
only.

[binjin, 2015/01/29 10:44:10]

I would like to keep the auto fetching as a backup, since the push connection
might still be unreliable with bad network connection. Maybe we can make the
interval to 3 hours as the same as command expiration time, in order to reduce
the server load. 

[bartfab (slow), 2015/01/29 15:28:58]

The purpose of our invalidation work was to significantly reduce server load as
we only talk to the server when there is actually something to do. Polling for
commands every 3 hours would add a lot of load. Also, I am not at all convinced
even a 3 hour interval would produce reasonable results. If pushing really is
broken, will an admin be OK waiting 3 hours for a command to execute? Is this
not an error state that requires escalation and a fix to pushing?

We should discuss this at our meeting with the server team tonight.

+
+}  // namespace
+
+namespace policy {
+
+namespace em = enterprise_management;
+
+RemoteCommandsService::RemoteCommandsService(
+    scoped_ptr<RemoteCommandJob::Factory> factory,
+    CloudPolicyClient* client)
+    : request_job_in_progress_(false),
+      unhandled_fetch_requests(0),
+      last_fetched_command_id_(0),
+      last_executed_command_id_(0),
+      factory_(factory.Pass()),
+      client_(client) {
+  queue_.AddObserver(this);
+}
+
+RemoteCommandsService::~RemoteCommandsService() {
+  queue_.RemoveObserver(this);
+}
+
+void RemoteCommandsService::FetchRemoteCommands() {
+  if (request_job_in_progress_ &&
+      unhandled_fetch_requests < kMaxUnhandledFetchRequests) {

[bartfab (slow), 2015/01/29 10:18:05]

Why do we need a queue of unhandled fetch requests? We could do things
analogously to policy: Only the most recent fetch request is sent to the server.
If there is a request in flight already, that request is dropped.

[binjin, 2015/01/29 10:44:10]

The entire logic here is for reliability. There is one assumption that I'm not
100% sure of in every corner cases and possibly unstable state of browser:
DeviceManagementService will eventually call
RemoteCommandsService::OnRemoteCommandsFetched() in reasonable time. After
kMaxUnhandledFetchRequests is reached we will simply cancel the current
connection job since it's either literally timeout or have a lot of other
pending fetching requets.

[bartfab (slow), 2015/01/29 15:28:58]

CloudPolicyClient keeps track of timeouts on its own. If a request to the server
really gets stuck (e.g. due to packet loss), it will time out and you will be
notified. You should resend the request in that case.

I really see no reason to keep a queue of requests here. There should only ever
be one request.

+    unhandled_fetch_requests++;
+    return;
+  }
+
+  request_job_in_progress_ = true;
+  unhandled_fetch_requests = 0;
+
+  request_job_ = client_->CreateRemoteCommandsJob();
+
+  em::DeviceRemoteCommandRequest* request =
+      request_job_->GetRequest()->mutable_remote_command_request();
+
+  request->set_last_command_unique_id(last_fetched_command_id_);
+  request->clear_command_results();
+  for (const auto& command_result : unsent_results_)

[bartfab (slow), 2015/01/29 10:18:05]

This has an interesting interaction with the reboot command, and with reboots in
general: Since we do not store unsent results in persistent storage, we should
try hard to flush any unsent results before rebooting.

[binjin, 2015/01/29 10:44:10]

Yes, I will make necessary change.

+    *request->add_command_results() = command_result;
+  unsent_results_.clear();
+
+  request_job_->Start(base::Bind(
+      &RemoteCommandsService::OnRemoteCommandsFetched, base::Unretained(this)));
+}
+
+void RemoteCommandsService::StartAutoFetching() {
+  timer_.Start(FROM_HERE,
+               base::TimeDelta::FromMinutes(kAutoFetchingIntervalInMinutes),
+               this, &RemoteCommandsService::FetchRemoteCommands);
+}
+
+void RemoteCommandsService::StopAutoFetching() {
+  timer_.Stop();
+}
+
+void RemoteCommandsService::AddCommand(
+    const enterprise_management::RemoteCommand& command) {
+  if (!command.has_type() || !command.has_unique_id()) {
+    LOG(WARNING) << "Invalid remote command from server.";
+    return;
+  }
+
+  last_fetched_command_id_ =

[bartfab (slow), 2015/01/29 10:18:05]

This means that we auto-acknowledge the command with the next fetch. That is not
quite what we want to do: As long as the command has not been processed, we
should *not* acknowledge its ID. We should allow the server to send it again and
again with subsequent fetches. That way, we can make sure the command is still
available to us e.g. after a crash or reboot.

But note that this also means we can get the same command from the server
multiple times. We need to account for this and ensure that we do not push
command duplicates into the queue.

[binjin, 2015/01/29 10:44:10]

Actually this patchset considered "RemoteCommandsService::OnJobStarted()" as an
acknowledge by writing |last_executed_command_id_| to a persistent source. So
that ideally after OnJobStarted() is called, we won't execute this commands
again.

I think I will make following changes:
* keep both |last_executed_command_id_| and |last_fetched_command_id_| recorded
in the same way as now
* send |last_executed_command_id_| to server for fetching instead.
* compare the fetched commands with |last_fetched_command_id_| to drop existing
ones
* (possibly) do not write |last_executed_command_id_| to persistent source now

[bartfab (slow), 2015/01/29 15:28:58]

I think this plan is heading in the right direction. We can discuss the details
once you have a prototype implementation of it.

+      std::max(last_fetched_command_id_, command.unique_id());
+
+  scoped_ptr<RemoteCommandJob> job = factory_->BuildJobForType(command.type());
+
+  if (!job || !job->Init(command)) {
+    em::RemoteCommandResult ignored_result;
+    ignored_result.set_result(
+        em::RemoteCommandResult_ResultType_RESULT_IGNORED);
+    ignored_result.set_unique_id(command.unique_id());
+    unsent_results_.push_back(ignored_result);
+    unhandled_fetch_requests++;
+    return;
+  }
+
+  queue_.AddJob(job.Pass());
+}
+
+void RemoteCommandsService::OnJobStarted(RemoteCommandJob* command) {
+  last_executed_command_id_ =
+      std::max(last_executed_command_id_, command->unique_id());
+  // XXX: sync |last_executed_command_id_| to persistent source, and reload
+  //      later.
+}
+
+void RemoteCommandsService::OnJobFinished(RemoteCommandJob* command) {
+  em::RemoteCommandResult result;
+  result.set_unique_id(command->unique_id());
+  result.set_timestamp((command->execution_started_time() -
+                        base::Time::UnixEpoch()).InMilliseconds());
+
+  if (command->status() == RemoteCommandJob::SUCCEEDED) {
+    result.set_result(em::RemoteCommandResult_ResultType_RESULT_SUCCESS);
+    scoped_ptr<std::string> result_payload = command->GetResultPayload();
+    if (result_payload)
+      result.set_payload(*result_payload);
+  } else if (command->status() == RemoteCommandJob::EXPIRED ||
+             command->status() == RemoteCommandJob::INVALID) {
+    result.set_result(em::RemoteCommandResult_ResultType_RESULT_IGNORED);
+  } else {
+    result.set_result(em::RemoteCommandResult_ResultType_RESULT_FAILURE);
+  }
+
+  unsent_results_.push_back(result);
+  unhandled_fetch_requests++;
+
+  FetchRemoteCommands();
+}
+
+void RemoteCommandsService::OnRemoteCommandsFetched(
+    DeviceManagementStatus status,
+    int net_error,
+    const em::DeviceManagementResponse& response) {
+  request_job_in_progress_ = false;
+
+  if (status == DM_STATUS_SUCCESS && response.has_remote_command_response()) {
+    for (const auto& command : response.remote_command_response().commands())
+      AddCommand(command);
+  }
+
+  if (unhandled_fetch_requests > 0)
+    FetchRemoteCommands();
+}
+
+}  // namespace policy