text: // Licensed to the Apache Software Foundation (ASF) under one
  // or more contributor license agreements.  See the NOTICE file
  // distributed with this work for additional information
  // regarding copyright ownership.  The ASF licenses this file
  // to you under the Apache License, Version 2.0 (the
  // "License"); you may not use this file except in compliance
  // with the License.  You may obtain a copy of the License at
  //
  //     http://www.apache.org/licenses/LICENSE-2.0
  //
  // Unless required by applicable law or agreed to in writing, software
  // distributed under the License is distributed on an "AS IS" BASIS,
  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  // See the License for the specific language governing permissions and
  // limitations under the License.
  
  #include <deque>
  #include <string>
  
  #include <mesos/type_utils.hpp>
  
  #include <mesos/state/state.hpp>
  
  #include <process/defer.hpp>
  #include <process/dispatch.hpp>
  #include <process/future.hpp>
  #include <process/help.hpp>
  #include <process/http.hpp>
  #include <process/id.hpp>
  #include <process/owned.hpp>
  #include <process/process.hpp>
  
  #include <process/metrics/gauge.hpp>
  #include <process/metrics/metrics.hpp>
  #include <process/metrics/timer.hpp>
  
  #include <stout/lambda.hpp>
  #include <stout/none.hpp>
  #include <stout/nothing.hpp>
  #include <stout/option.hpp>
  #include <stout/protobuf.hpp>
  #include <stout/stopwatch.hpp>
  
  #include "master/registrar.hpp"
  #include "master/registry.hpp"
  
  using mesos::state::State;
  using mesos::state::Variable;
  
  using process::dispatch;
  using process::spawn;
  using process::terminate;
  using process::wait; // Necessary on some OS's to disambiguate.
  
  using process::AUTHENTICATION;
  using process::DESCRIPTION;
  using process::Failure;
  using process::Future;
  using process::HELP;
  using process::Owned;
  using process::PID;
  using process::Process;
  using process::Promise;
  using process::TLDR;
  
  using process::http::OK;
  
  using process::http::authentication::Principal;
  
  using process::metrics::Gauge;
  using process::metrics::Timer;
  
  using std::deque;
  using std::string;
  
  namespace mesos {
  namespace internal {
  namespace master {
  
  using process::http::Response;
  using process::http::Request;
  
  class RegistrarProcess : public Process<RegistrarProcess>
  {
  public:
    RegistrarProcess(
        const Flags& _flags,
        State* _state,
        const Option<string>& _authenticationRealm)
      : ProcessBase(process::ID::generate("registrar")),
        metrics(*this),
        state(_state),
        updating(false),
        flags(_flags),
        authenticationRealm(_authenticationRealm) {}
  
    virtual ~RegistrarProcess() {}
  
    // Registrar implementation.
    Future<Registry> recover(const MasterInfo& info);
    Future<bool> apply(Owned<Operation> operation);
  
  protected:
    virtual void initialize()
    {
      if (authenticationRealm.isSome()) {
        route(
            "/registry",
            authenticationRealm.get(),
            registryHelp(),
            &RegistrarProcess::getRegistry);
      } else {
        route(
            "/registry",
            registryHelp(),
            lambda::bind(
                &RegistrarProcess::getRegistry, this, lambda::_1, None()));
      }
    }
  
  private:
    // HTTP handlers.
    // /registrar(N)/registry
    Future<Response> getRegistry(
        const Request& request,
        const Option<Principal>&);
    static string registryHelp();
  
    // The 'Recover' operation adds the latest MasterInfo.
    class Recover : public Operation
    {
    public:
      explicit Recover(const MasterInfo& _info) : info(_info) {}
  
    protected:
      virtual Try<bool> perform(Registry* registry, hashset<SlaveID>* slaveIDs)
      {
        registry->mutable_master()->mutable_info()->CopyFrom(info);
        return true; // Mutation.
      }
  
    private:
      const MasterInfo info;
    };
  
    // Metrics.
    struct Metrics
    {
      explicit Metrics(const RegistrarProcess& process)
        : queued_operations(
              "registrar/queued_operations",
              defer(process, &RegistrarProcess::_queued_operations)),
          registry_size_bytes(
              "registrar/registry_size_bytes",
              defer(process, &RegistrarProcess::_registry_size_bytes)),
          state_fetch("registrar/state_fetch"),
          state_store("registrar/state_store", Days(1))
      {
        process::metrics::add(queued_operations);
        process::metrics::add(registry_size_bytes);
  
        process::metrics::add(state_fetch);
        process::metrics::add(state_store);
      }
  
      ~Metrics()
      {
        process::metrics::remove(queued_operations);
        process::metrics::remove(registry_size_bytes);
  
        process::metrics::remove(state_fetch);
        process::metrics::remove(state_store);
      }
  
      Gauge queued_operations;
      Gauge registry_size_bytes;
  
      Timer<Milliseconds> state_fetch;
      Timer<Milliseconds> state_store;
    } metrics;
  
    // Gauge handlers.
    double _queued_operations()
    {
      return operations.size();
    }
  
    Future<double> _registry_size_bytes()
    {
      if (registry.isSome()) {
        return registry->ByteSize();
      }
  
      return Failure("Not recovered yet");
    }
  
    // Continuations.
    void _recover(
        const MasterInfo& info,
        const Future<Variable>& recovery);
    void __recover(const Future<bool>& recover);
    Future<bool> _apply(Owned<Operation> operation);
  
    // Helper for updating state (performing store).
    void update();
    void _update(
        const Future<Option<Variable>>& store,
        const Owned<Registry>& updatedRegistry,
        deque<Owned<Operation>> operations);
  
    // Fails all pending operations and transitions the Registrar
    // into an error state in which all subsequent operations will fail.
    // This ensures we don't attempt to re-acquire log leadership by
    // performing more State storage operations.
    void abort(const string& message);
  
    // TODO(ipronin): We use the "untyped" `State` class here and perform
    // the protobuf (de)serialization manually within the Registrar, because
    // the use of `protobuf::State` incurs a dramatic peformance cost from
    // protobuf copying. We should explore using `protobuf::State`, which will
    // require move support and other copy elimination to maintain the
    // performance of the current approach.
    State* state;
  
    // Per the TODO above, we store both serialized and deserialized versions
    // of the `Registry` protobuf. If we're able to move to `protobuf::State`,
    // we could just store a single `protobuf::state::Variable<Registry>`.
    Option<Variable> variable;
    Option<Registry> registry;
  
    deque<Owned<Operation>> operations;
    bool updating; // Used to signify fetching (recovering) or storing.
  
    const Flags flags;
  
    // Used to compose our operations with recovery.
    Option<Owned<Promise<Registry>>> recovered;
  
    // When an error is encountered from abort(), we'll fail all
    // subsequent operations.
    Option<Error> error;
  
    // The authentication realm, if any, into which this process'
    // endpoints will be installed.
    Option<string> authenticationRealm;
  };
  
  
  // Helper for treating State operations that timeout as failures.
  template <typename T>
  Future<T> timeout(
      const string& operation,
      const Duration& duration,
      Future<T> future)
  {
    future.discard();
  
    return Failure(
        "Failed to perform " + operation + " within " + stringify(duration));
  }
  
  
  // Helper for failing a deque of operations.
  void fail(deque<Owned<Operation>>* operations, const string& message)
  {
    while (!operations->empty()) {
      operations->front()->fail(message);
      operations->pop_front();
    }
  }
  
  
  Future<Response> RegistrarProcess::getRegistry(
      const Request& request,
      const Option<Principal>&)
  {
    JSON::Object result;
  
    if (registry.isSome()) {
      result = JSON::protobuf(registry.get());
    }
  
    return OK(result, request.url.query.get("jsonp"));
  }
  
  
  string RegistrarProcess::registryHelp()
  {
    return HELP(
        TLDR(
            "Returns the current contents of the Registry in JSON."),
        DESCRIPTION(
            "Example:",
            "",
            "```",
            "{",
            "  \"master\":",
            "  {",
            "    \"info\":",
            "    {",
            "      \"hostname\": \"localhost\",",
            "      \"id\": \"20140325-235542-1740121354-5050-33357\",",
            "      \"ip\": 2130706433,",
            "      \"pid\": \"master@127.0.0.1:5050\",",
            "      \"port\": 5050",
            "    }",
            "  },",
            "",
            "  \"slaves\":",
            "  {",
            "    \"slaves\":",
            "    [",
            "      {",
            "        \"info\":",
            "        {",
            "          \"checkpoint\": true,",
            "          \"hostname\": \"localhost\",",
            "          \"id\":",
            "          {",
            "            \"value\": \"20140325-234618-1740121354-5050-29065-0\"",
            "          },",
            "          \"port\": 5051,",
            "          \"resources\":",
            "          [",
            "            {",
            "              \"name\": \"cpus\",",
            "              \"role\": \"*\",",
            "              \"scalar\": { \"value\": 24 },",
            "              \"type\": \"SCALAR\"",
            "            }",
            "          ],",
            "        }",
            "      }",
            "    ]",
            "  }",
            "}",
            "```"),
        AUTHENTICATION(true));
  }
  
  
  Future<Registry> RegistrarProcess::recover(const MasterInfo& info)
  {
    if (recovered.isNone()) {
      VLOG(1) << "Recovering registrar";
  
      metrics.state_fetch.start();
      state->fetch("registry")
        .after(flags.registry_fetch_timeout,
               lambda::bind(
                   &timeout<Variable>,
                   "fetch",
                   flags.registry_fetch_timeout,
                   lambda::_1))
        .onAny(defer(self(), &Self::_recover, info, lambda::_1));
      updating = true;
      recovered = Owned<Promise<Registry>>(new Promise<Registry>());
    }
  
    return recovered.get()->future();
  }
  
  
  void RegistrarProcess::_recover(
      const MasterInfo& info,
      const Future<Variable>& recovery)
  {
    updating = false;
  
    CHECK(!recovery.isPending());
  
    if (!recovery.isReady()) {
      recovered.get()->fail("Failed to recover registrar: " +
          (recovery.isFailed() ? recovery.failure() : "discarded"));
      return;
    }
  
    // Deserialize the registry.
    Try<Registry> deserialized =
      ::protobuf::deserialize<Registry>(recovery->value());
    if (deserialized.isError()) {
      recovered.get()->fail("Failed to recover registrar: " +
                            deserialized.error());
      return;
    }
  
    Duration elapsed = metrics.state_fetch.stop();
  
    LOG(INFO) << "Successfully fetched the registry"
              << " (" << Bytes(deserialized->ByteSize()) << ")"
              << " in " << elapsed;
  
    // Save the registry.
    variable = recovery.get();
  
    // Workaround for immovable protobuf messages.
    registry = Option<Registry>(Registry());
    registry->Swap(&deserialized.get());
  
    // Perform the Recover operation to add the new MasterInfo.
    Owned<Operation> operation(new Recover(info));
    operations.push_back(operation);
    operation->future()
      .onAny(defer(self(), &Self::__recover, lambda::_1));
  
    update();
  }
  
  
  void RegistrarProcess::__recover(const Future<bool>& recover)
  {
    CHECK(!recover.isPending());
  
    if (!recover.isReady()) {
      recovered.get()->fail("Failed to recover registrar: "
          "Failed to persist MasterInfo: " +
          (recover.isFailed() ? recover.failure() : "discarded"));
    } else if (!recover.get()) {
      recovered.get()->fail("Failed to recover registrar: "
          "Failed to persist MasterInfo: version mismatch");
    } else {
      LOG(INFO) << "Successfully recovered registrar";
  
      // At this point _update() has updated 'variable' to contain
      // the Registry with the latest MasterInfo.
      // Set the promise and un-gate any pending operations.
      CHECK_SOME(variable);
      CHECK_SOME(registry);
      recovered.get()->set(registry.get());
    }
  }
  
  
  Future<bool> RegistrarProcess::apply(Owned<Operation> operation)
  {
    if (recovered.isNone()) {
      return Failure("Attempted to apply the operation before recovering");
    }
  
    return recovered.get()->future()
      .then(defer(self(), &Self::_apply, operation));
  }
  
  
  Future<bool> RegistrarProcess::_apply(Owned<Operation> operation)
  {
    if (error.isSome()) {
      return Failure(error.get());
    }
  
    CHECK_SOME(variable);
  
    operations.push_back(operation);
    Future<bool> future = operation->future();
    if (!updating) {
      update();
    }
    return future;
  }
  
  
  void RegistrarProcess::update()
  {
    if (operations.empty()) {
      return; // No-op.
    }
  
    CHECK(!updating);
    CHECK_NONE(error);
    CHECK_SOME(variable);
  
    // Time how long it takes to apply the operations.
    Stopwatch stopwatch;
    stopwatch.start();
  
    updating = true;
  
    // Create a snapshot of the current registry. We use an `Owned` here
    // to avoid copying, since protobuf doesn't suppport move construction.
    auto updatedRegistry = Owned<Registry>(new Registry(registry.get()));
  
    // Create the 'slaveIDs' accumulator.
    hashset<SlaveID> slaveIDs;
    foreach (const Registry::Slave& slave, updatedRegistry->slaves().slaves()) {
      slaveIDs.insert(slave.info().id());
    }
  
    foreach (Owned<Operation>& operation, operations) {
      // No need to process the result of the operation.
      (*operation)(updatedRegistry.get(), &slaveIDs);
    }
  
    LOG(INFO) << "Applied " << operations.size() << " operations in "
              << stopwatch.elapsed() << "; attempting to update the registry";
  
    // Perform the store, and time the operation.
    metrics.state_store.start();
  
    // Serialize updated registry.
    Try<string> serialized = ::protobuf::serialize(*updatedRegistry);
    if (serialized.isError()) {
      string message = "Failed to update registry: " + serialized.error();
      fail(&operations, message);
      abort(message);
      return;
    }
  
    state->store(variable->mutate(serialized.get()))
      .after(flags.registry_store_timeout,
             lambda::bind(
                 &timeout<Option<Variable>>,
                 "store",
                 flags.registry_store_timeout,
                 lambda::_1))
      .onAny(defer(
          self(), &Self::_update, lambda::_1, updatedRegistry, operations));
  
    // Clear the operations, _update will transition the Promises!
    operations.clear();
  }
  
  
  void RegistrarProcess::_update(
      const Future<Option<Variable>>& store,
      const Owned<Registry>& updatedRegistry,
      deque<Owned<Operation>> applied)
  {
    updating = false;
  
    // Abort if the storage operation did not succeed.
    if (!store.isReady() || store.get().isNone()) {
      string message = "Failed to update registry: ";
  
      if (store.isFailed()) {
        message += store.failure();
      } else if (store.isDiscarded()) {
        message += "discarded";
      } else {
        message += "version mismatch";
      }
  
      fail(&applied, message);
      abort(message);
  
      return;
    }
  
    Duration elapsed = metrics.state_store.stop();
  
    LOG(INFO) << "Successfully updated the registry in " << elapsed;
  
    variable = store.get().get();
    registry->Swap(updatedRegistry.get());
  
    // Remove the operations.
    while (!applied.empty()) {
      Owned<Operation> operation = applied.front();
      applied.pop_front();
  
      operation->set();
    }
  
    if (!operations.empty()) {
      update();
    }
  }
  
  
  void RegistrarProcess::abort(const string& message)
  {
    error = Error(message);
  
    LOG(ERROR) << "Registrar aborting: " << message;
  
    fail(&operations, message);
  }
  
  
  Registrar::Registrar(
      const Flags& flags,
      State* state,
      const Option<string>& authenticationRealm)
  {
    process = new RegistrarProcess(flags, state, authenticationRealm);
    spawn(process);
  }
  
  
  Registrar::~Registrar()
  {
    terminate(process);
    wait(process);
    delete process;
  }
  
  
  Future<Registry> Registrar::recover(const MasterInfo& info)
  {
    return dispatch(process, &RegistrarProcess::recover, info);
  }
  
  
  Future<bool> Registrar::apply(Owned<Operation> operation)
  {
    return dispatch(process, &RegistrarProcess::apply, operation);
  }
  
  PID<RegistrarProcess> Registrar::pid() const
  {
    return process->self();
  }
  
  } // namespace master {
  } // namespace internal {
  } // namespace mesos {
  