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 <algorithm>
  #include <list>
  #include <memory>
  #include <set>
  #include <string>
  #include <tuple>
  #include <vector>
  
  #include <google/protobuf/repeated_field.h>
  
  #include <mesos/csi/v0.hpp>
  #include <mesos/csi/v1.hpp>
  
  #include <process/clock.hpp>
  #include <process/collect.hpp>
  #include <process/future.hpp>
  #include <process/http.hpp>
  #include <process/grpc.hpp>
  #include <process/gtest.hpp>
  #include <process/gmock.hpp>
  #include <process/protobuf.hpp>
  #include <process/queue.hpp>
  #include <process/reap.hpp>
  
  #include <stout/foreach.hpp>
  #include <stout/hashmap.hpp>
  #include <stout/stringify.hpp>
  #include <stout/strings.hpp>
  #include <stout/unreachable.hpp>
  #include <stout/uri.hpp>
  #include <stout/uuid.hpp>
  
  #include <stout/os/realpath.hpp>
  
  #include "csi/constants.hpp"
  #include "csi/paths.hpp"
  #include "csi/state.hpp"
  #include "csi/v0_volume_manager_process.hpp"
  #include "csi/v1_volume_manager_process.hpp"
  #include "csi/volume_manager.hpp"
  
  #include "linux/fs.hpp"
  
  #include "master/detector/standalone.hpp"
  
  #include "module/manager.hpp"
  
  #include "messages/messages.hpp"
  
  #include "resource_provider/constants.hpp"
  
  #include "slave/container_daemon_process.hpp"
  #include "slave/paths.hpp"
  #include "slave/state.hpp"
  
  #include "slave/containerizer/fetcher.hpp"
  
  #include "slave/containerizer/mesos/containerizer.hpp"
  
  #include "status_update_manager/status_update_manager_process.hpp"
  
  #include "tests/disk_profile_server.hpp"
  #include "tests/environment.hpp"
  #include "tests/flags.hpp"
  #include "tests/mesos.hpp"
  #include "tests/mock_csi_plugin.hpp"
  
  namespace http = process::http;
  
  using std::list;
  using std::multiset;
  using std::shared_ptr;
  using std::string;
  using std::vector;
  
  using google::protobuf::RepeatedPtrField;
  
  using mesos::internal::slave::ContainerDaemonProcess;
  
  using mesos::master::detector::MasterDetector;
  using mesos::master::detector::StandaloneMasterDetector;
  
  using process::Clock;
  using process::Future;
  using process::Owned;
  using process::Promise;
  using process::post;
  using process::Queue;
  using process::reap;
  
  using process::grpc::StatusError;
  
  using testing::_;
  using testing::A;
  using testing::AllOf;
  using testing::AtMost;
  using testing::Between;
  using testing::DoAll;
  using testing::Not;
  using testing::Sequence;
  
  namespace mesos {
  namespace internal {
  namespace tests {
  
  constexpr char URI_DISK_PROFILE_ADAPTOR_NAME[] =
    "org_apache_mesos_UriDiskProfileAdaptor";
  
  constexpr char TEST_SLRP_TYPE[] = "org.apache.mesos.rp.local.storage";
  constexpr char TEST_SLRP_NAME[] = "test";
  constexpr char TEST_CSI_PLUGIN_TYPE[] = "org.apache.mesos.csi.test";
  constexpr char TEST_CSI_PLUGIN_NAME[] = "local";
  constexpr char TEST_CSI_VENDOR[] = "org.apache.mesos.csi.test.local";
  
  
  class StorageLocalResourceProviderTest
    : public ContainerizerTest<slave::MesosContainerizer>,
      public testing::WithParamInterface<string>
  {
  public:
    void SetUp() override
    {
      ContainerizerTest<slave::MesosContainerizer>::SetUp();
  
      resourceProviderConfigDir =
        path::join(sandbox.get(), "resource_provider_configs");
  
      ASSERT_SOME(os::mkdir(resourceProviderConfigDir.get()));
  
      Try<string> mkdtemp = environment->mkdtemp();
      ASSERT_SOME(mkdtemp);
  
      testCsiPluginWorkDir = mkdtemp.get();
    }
  
    void TearDown() override
    {
      // Unload modules.
      foreach (const Modules::Library& library, modules.libraries()) {
        foreach (const Modules::Library::Module& module, library.modules()) {
          if (module.has_name()) {
            ASSERT_SOME(modules::ModuleManager::unload(module.name()));
          }
        }
      }
  
      foreach (const string& slaveWorkDir, slaveWorkDirs) {
        // Clean up CSI endpoint directories if there is any.
        const string csiRootDir = slave::paths::getCsiRootDir(slaveWorkDir);
  
        Try<list<string>> csiContainerPaths =
          csi::paths::getContainerPaths(csiRootDir, "*", "*");
        ASSERT_SOME(csiContainerPaths);
  
        foreach (const string& path, csiContainerPaths.get()) {
          Try<csi::paths::ContainerPath> containerPath =
            csi::paths::parseContainerPath(csiRootDir, path);
          ASSERT_SOME(containerPath);
  
          Result<string> endpointDir =
            os::realpath(csi::paths::getEndpointDirSymlinkPath(
                csiRootDir,
                containerPath->type,
                containerPath->name,
                containerPath->containerId));
  
          if (endpointDir.isSome()) {
            ASSERT_SOME(os::rmdir(endpointDir.get()));
          }
        }
      }
  
      ContainerizerTest<slave::MesosContainerizer>::TearDown();
    }
  
    master::Flags CreateMasterFlags() override
    {
      master::Flags flags =
        ContainerizerTest<slave::MesosContainerizer>::CreateMasterFlags();
  
      // Use a small allocation interval to speed up the test. We do this instead
      // of manipulating the clock because the storage local resource provider
      // relies on a running clock to wait for the CSI plugin to be ready.
      flags.allocation_interval = Milliseconds(50);
  
      return flags;
    }
  
    slave::Flags CreateSlaveFlags() override
    {
      slave::Flags flags =
        ContainerizerTest<slave::MesosContainerizer>::CreateSlaveFlags();
  
      flags.resource_provider_config_dir = resourceProviderConfigDir;
  
      // Store the agent work directory for cleaning up CSI endpoint
      // directories during teardown.
      // NOTE: DO NOT change the work directory afterward.
      slaveWorkDirs.push_back(flags.work_dir);
  
      return flags;
    }
  
    void loadUriDiskProfileAdaptorModule(
        const string& uri,
        const Option<Duration> pollInterval = None())
    {
      const string libraryPath = getModulePath("uri_disk_profile_adaptor");
  
      Modules::Library* library = modules.add_libraries();
      library->set_name("uri_disk_profile_adaptor");
      library->set_file(libraryPath);
  
      Modules::Library::Module* module = library->add_modules();
      module->set_name(URI_DISK_PROFILE_ADAPTOR_NAME);
  
      Parameter* _uri = module->add_parameters();
      _uri->set_key("uri");
      _uri->set_value(uri);
  
      if (pollInterval.isSome()) {
        Parameter* _pollInterval = module->add_parameters();
        _pollInterval->set_key("poll_interval");
        _pollInterval->set_value(stringify(pollInterval.get()));
      }
  
      ASSERT_SOME(modules::ModuleManager::load(modules));
    }
  
    void setupResourceProviderConfig(
        const Bytes& capacity,
        const Option<string>& volumes = None(),
        const Option<string>& forward = None(),
        const Option<string>& createParameters = None(),
        const Option<string>& volumeMetadata = None(),
        const Option<Duration>& reconciliationInterval = None())
    {
      const string testCsiPluginPath =
        path::join(tests::flags.build_dir, "src", "test-csi-plugin");
  
      Try<string> resourceProviderConfig = strings::format(
          R"~(
          {
            "type": "%s",
            "name": "%s",
            "default_reservations": [
              {
                "type": "DYNAMIC",
                "role": "storage"
              }
            ],
            "storage": {
              "plugin": {
                "type": "%s",
                "name": "%s",
                "containers": [
                  {
                    "services": [
                      "CONTROLLER_SERVICE",
                      "NODE_SERVICE"
                    ],
                    "command": {
                      "shell": false,
                      "value": "%s",
                      "arguments": [
                        "%s",
                        "--api_version=%s",
                        "--work_dir=%s",
                        "--available_capacity=%s",
                        "--volumes=%s",
                        "%s",
                        "--create_parameters=%s",
                        "--volume_metadata=%s"
                      ]
                    },
                    "resources": [
                      {
                        "name": "cpus",
                        "type": "SCALAR",
                        "scalar": {
                          "value": 0.1
                        }
                      },
                      {
                        "name": "mem",
                        "type": "SCALAR",
                        "scalar": {
                          "value": 1024
                        }
                      }
                    ]
                  }
                ]
              },
              "reconciliation_interval_seconds" : %s
            }
          }
          )~",
          TEST_SLRP_TYPE,
          TEST_SLRP_NAME,
          TEST_CSI_PLUGIN_TYPE,
          TEST_CSI_PLUGIN_NAME,
          testCsiPluginPath,
          testCsiPluginPath,
          GetParam(),
          testCsiPluginWorkDir.get(),
          stringify(capacity),
          volumes.getOrElse(""),
          forward.isSome() ? "--forward=" + forward.get() : "",
          createParameters.getOrElse(""),
          volumeMetadata.getOrElse(""),
          stringify(reconciliationInterval.getOrElse(Seconds(0)).secs()));
  
      ASSERT_SOME(resourceProviderConfig);
  
      ASSERT_SOME(os::write(
          path::join(resourceProviderConfigDir.get(), "test.json"),
          resourceProviderConfig.get()));
    }
  
    // Set up an expected `CreateVolume` CSI call for a given mock CSI plugin.
    // When the call is made to the mock plugin, `result` will be responded. When
    // the response is received by the volume manager, the returned future will be
    // satisfied.
    Future<Nothing> futureCreateVolumeCall(
        MockCSIPlugin* plugin, const Try<csi::VolumeInfo, StatusError>& result)
    {
      if (GetParam() == csi::v0::API_VERSION) {
        EXPECT_CALL(*plugin, CreateVolume(
            _, _, A<csi::v0::CreateVolumeResponse*>()))
          .WillOnce(Invoke([result](
              grpc::ServerContext* context,
              const csi::v0::CreateVolumeRequest* request,
              csi::v0::CreateVolumeResponse* response) {
            if (result.isError()) {
              return result.error().status;
            }
  
            response->mutable_volume()->set_id(result->id);
            response->mutable_volume()
              ->set_capacity_bytes(result->capacity.bytes());
            *response->mutable_volume()->mutable_attributes() = result->context;
  
            return grpc::Status::OK;
          }));
  
        return FUTURE_DISPATCH(_, &csi::v0::VolumeManagerProcess::__call<
            csi::v0::CreateVolumeResponse>);
      } else if (GetParam() == csi::v1::API_VERSION) {
        EXPECT_CALL(*plugin, CreateVolume(
            _, _, A<csi::v1::CreateVolumeResponse*>()))
          .WillOnce(Invoke([result](
              grpc::ServerContext* context,
              const csi::v1::CreateVolumeRequest* request,
              csi::v1::CreateVolumeResponse* response) {
            if (result.isError()) {
              return result.error().status;
            }
  
            response->mutable_volume()->set_volume_id(result->id);
            response->mutable_volume()
              ->set_capacity_bytes(result->capacity.bytes());
            *response->mutable_volume()
              ->mutable_volume_context() = result->context;
  
            return grpc::Status::OK;
          }));
  
        return FUTURE_DISPATCH(_, &csi::v1::VolumeManagerProcess::__call<
            csi::v1::CreateVolumeResponse>);
      }
  
      // This extra closure is necessary in order to use `FAIL` as it requires a
      // void return type.
      [&] { FAIL() << "Unsupported CSI API version " << GetParam(); }();
  
      UNREACHABLE();
    }
  
    // Set up an expected `DeleteVolume` CSI call for a given mock CSI plugin.
    // When the call is made to the mock plugin, `result` will be responded. When
    // the response is received by the volume manager, the returned future will be
    // satisfied.
    Future<Nothing> futureDeleteVolumeCall(
        MockCSIPlugin* plugin, const Try<Nothing, StatusError>& result)
    {
      if (GetParam() == csi::v0::API_VERSION) {
        EXPECT_CALL(*plugin, DeleteVolume(
            _, _, A<csi::v0::DeleteVolumeResponse*>()))
          .WillOnce(Invoke([result](
              grpc::ServerContext* context,
              const csi::v0::DeleteVolumeRequest* request,
              csi::v0::DeleteVolumeResponse* response) {
            return result.isError() ? result.error().status : grpc::Status::OK;
          }));
  
        return FUTURE_DISPATCH(_, &csi::v0::VolumeManagerProcess::__call<
            csi::v0::DeleteVolumeResponse>);
      } else if (GetParam() == csi::v1::API_VERSION) {
        EXPECT_CALL(*plugin, DeleteVolume(
            _, _, A<csi::v1::DeleteVolumeResponse*>()))
          .WillOnce(Invoke([result](
              grpc::ServerContext* context,
              const csi::v1::DeleteVolumeRequest* request,
              csi::v1::DeleteVolumeResponse* response) {
            return result.isError() ? result.error().status : grpc::Status::OK;
          }));
  
        return FUTURE_DISPATCH(_, &csi::v1::VolumeManagerProcess::__call<
            csi::v1::DeleteVolumeResponse>);
      }
  
      // This extra closure is necessary in order to use `FAIL` as it requires a
      // void return type.
      [&] { FAIL() << "Unsupported CSI API version " << GetParam(); }();
  
      UNREACHABLE();
    }
  
    // Set up an expected `NodePublishVolume` CSI call for a given mock CSI
    // plugin. When the call is made to the mock plugin, `result` will be
    // responded. When the response is received by the volume manager, the
    // returned future will be satisfied.
    Future<Nothing> futureNodePublishVolumeCall(
        MockCSIPlugin* plugin, const Try<Nothing, StatusError>& result)
    {
      if (GetParam() == csi::v0::API_VERSION) {
        EXPECT_CALL(*plugin, NodePublishVolume(
            _, _, A<csi::v0::NodePublishVolumeResponse*>()))
          .WillOnce(Invoke([result](
              grpc::ServerContext* context,
              const csi::v0::NodePublishVolumeRequest* request,
              csi::v0::NodePublishVolumeResponse* response) {
            return result.isError() ? result.error().status : grpc::Status::OK;
          }));
  
        return FUTURE_DISPATCH(_, &csi::v0::VolumeManagerProcess::__call<
            csi::v0::NodePublishVolumeResponse>);
      } else if (GetParam() == csi::v1::API_VERSION) {
        EXPECT_CALL(*plugin, NodePublishVolume(
            _, _, A<csi::v1::NodePublishVolumeResponse*>()))
          .WillOnce(Invoke([result](
              grpc::ServerContext* context,
              const csi::v1::NodePublishVolumeRequest* request,
              csi::v1::NodePublishVolumeResponse* response) {
            return result.isError() ? result.error().status : grpc::Status::OK;
          }));
  
        return FUTURE_DISPATCH(_, &csi::v1::VolumeManagerProcess::__call<
            csi::v1::NodePublishVolumeResponse>);
      }
  
      // This extra closure is necessary in order to use `FAIL` as it requires a
      // void return type.
      [&] { FAIL() << "Unsupported CSI API version " << GetParam(); }();
  
      UNREACHABLE();
    }
  
    // Create a JSON string representing a disk profile mapping containing the
    // given profile-parameter pairs.
    static string createDiskProfileMapping(
        const hashmap<string, Option<JSON::Object>>& profiles)
    {
      JSON::Object diskProfileMapping{
        {"profile_matrix", JSON::Object{}}
      };
  
      foreachpair (const string& profile,
                   const Option<JSON::Object>& parameters,
                   profiles) {
        JSON::Object profileInfo{
          {"csi_plugin_type_selector", JSON::Object{
            {"plugin_type", "org.apache.mesos.csi.test"}
          }},
          {"volume_capabilities", JSON::Object{
            {"mount", JSON::Object{}},
            {"access_mode", JSON::Object{
              {"mode", "SINGLE_NODE_WRITER"}
            }}
          }}};
  
        diskProfileMapping
          .values.at("profile_matrix").as<JSON::Object>()
          .values.emplace(profile, profileInfo);
  
        if (parameters.isSome()) {
          diskProfileMapping
            .values.at("profile_matrix").as<JSON::Object>()
            .values.at(profile).as<JSON::Object>()
            .values.emplace("create_parameters", parameters.get());
        }
      }
  
      return stringify(diskProfileMapping);
    }
  
    static Resources createTaskResources(const Resources& additional)
    {
      return Resources::parse("cpus:0.1;mem:128").get() + additional;
    }
  
    static string metricName(const string& basename)
    {
      return "resource_providers/" + stringify(TEST_SLRP_TYPE) + "." +
        stringify(TEST_SLRP_NAME) + "/" + basename;
    }
  
  private:
    Modules modules;
    vector<string> slaveWorkDirs;
    Option<string> resourceProviderConfigDir;
    Option<string> testCsiPluginWorkDir;
  };
  
  
  // Tests whether a resource is a storage pool of a given profile. A storage pool
  // is a RAW disk resource with a profile but no source ID.
  template <typename Resource>
  static bool isStoragePool(const Resource& r, const string& profile)
  {
    return r.has_disk() &&
      r.disk().has_source() &&
      r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
      r.disk().source().has_vendor() &&
      r.disk().source().vendor() == TEST_CSI_VENDOR &&
      !r.disk().source().has_id() &&
      r.disk().source().has_profile() &&
      r.disk().source().profile() == profile;
  }
  
  
  // Tests whether a resource is a MOUNT disk of a given profile but not a
  // persistent volume. A MOUNT disk has both profile and source ID set.
  template <typename Resource>
  static bool isMountDisk(const Resource& r, const string& profile)
  {
    return r.has_disk() &&
      r.disk().has_source() &&
      r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
      r.disk().source().has_vendor() &&
      r.disk().source().vendor() == TEST_CSI_VENDOR &&
      r.disk().source().has_id() &&
      r.disk().source().has_profile() &&
      r.disk().source().profile() == profile &&
      !r.disk().has_persistence();
  }
  
  
  // Tests whether a resource is a BLOCK disk of a given profile but not a
  // persistent volume. A BLOCK disk has both profile and source ID set.
  template <typename Resource>
  static bool isBlockDisk(const Resource& r, const string& profile)
  {
    return r.has_disk() &&
      r.disk().has_source() &&
      r.disk().source().type() == Resource::DiskInfo::Source::BLOCK &&
      r.disk().source().has_vendor() &&
      r.disk().source().vendor() == TEST_CSI_VENDOR &&
      r.disk().source().has_id() &&
      r.disk().source().has_profile() &&
      r.disk().source().profile() == profile &&
      !r.disk().has_persistence();
  }
  
  
  // Tests whether a resource is a preprovisioned volume. A preprovisioned volume
  // is a RAW disk resource with a source ID but no profile.
  template <typename Resource>
  static bool isPreprovisionedVolume(const Resource& r)
  {
    return r.has_disk() &&
      r.disk().has_source() &&
      r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
      r.disk().source().has_vendor() &&
      r.disk().source().vendor() == TEST_CSI_VENDOR &&
      r.disk().source().has_id() &&
      !r.disk().source().has_profile();
  }
  
  
  INSTANTIATE_TEST_CASE_P(
      CSIVersion,
      StorageLocalResourceProviderTest,
      testing::Values(csi::v0::API_VERSION, csi::v1::API_VERSION),
      [](const testing::TestParamInfo<string>& info) { return info.param; });
  
  
  // This test verifies that a storage local resource provider can report
  // no resource and recover from this state.
  TEST_P(StorageLocalResourceProviderTest, NoResource)
  {
    Clock::pause();
  
    setupResourceProviderConfig(Bytes(0));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration and prevent retry.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
    EXPECT_EQ(
        0,
        updateSlave2->resource_providers().providers(0).total_resources_size());
  
    Clock::pause();
  
    // Restart the agent.
    slave.get()->terminate();
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave4 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave3 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration and prevent retry.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave3);
  
    Clock::resume();
  
    AWAIT_READY(updateSlave4);
    ASSERT_TRUE(updateSlave4->has_resource_providers());
    ASSERT_EQ(1, updateSlave4->resource_providers().providers_size());
    EXPECT_EQ(
        0,
        updateSlave4->resource_providers().providers(0).total_resources_size());
  }
  
  
  // This test verifies that any zero-sized volume reported by a CSI
  // plugin will be ignored by the storage local resource provider.
  TEST_P(StorageLocalResourceProviderTest, DISABLED_ZeroSizedDisk)
  {
    Clock::pause();
  
    setupResourceProviderConfig(Bytes(0), "volume0:0B");
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration and prevent retry.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Option<Resource> volume;
    foreach (const Resource& resource,
             updateSlave2->resource_providers().providers(0).total_resources()) {
      if (Resources::hasResourceProvider(resource)) {
        volume = resource;
      }
    }
  
    ASSERT_NONE(volume);
  }
  
  
  // This test verifies that the storage local resource provider can
  // handle disks less than 1MB correctly.
  TEST_P(StorageLocalResourceProviderTest, DISABLED_SmallDisk)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Kilobytes(512), "volume0:512KB");
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to receive offers.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    Future<vector<Offer>> rawDisksOffers;
  
    // We use the following filter to filter offers that do not have
    // wanted resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline offers that do not contain wanted resources.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // Since the resource provider always reports the pre-existing volume,
    // but only reports the storage pool after it gets the profile, an
    // offer containing the latter will also contain the former.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&rawDisksOffers));
  
    driver.start();
  
    AWAIT_READY(rawDisksOffers);
    ASSERT_FALSE(rawDisksOffers->empty());
  
    Option<Resource> storagePool;
    Option<Resource> preExistingVolume;
    foreach (const Resource& resource, rawDisksOffers->at(0).resources()) {
      if (isStoragePool(resource, "test")) {
        storagePool = resource;
      } else if (isPreprovisionedVolume(resource)) {
        preExistingVolume = resource;
      }
    }
  
    ASSERT_SOME(storagePool);
    EXPECT_EQ(
        Kilobytes(512),
        Bytes(storagePool->scalar().value() * Bytes::MEGABYTES));
  
    ASSERT_SOME(preExistingVolume);
    EXPECT_EQ(
        Kilobytes(512),
        Bytes(preExistingVolume->scalar().value() * Bytes::MEGABYTES));
  }
  
  
  // This test verifies that a framework can receive offers having new storage
  // pools from the storage local resource provider after a profile appears.
  TEST_P(StorageLocalResourceProviderTest, ProfileAppeared)
  {
    Clock::pause();
  
    Future<shared_ptr<TestDiskProfileServer>> server =
      TestDiskProfileServer::create();
    AWAIT_READY(server);
  
    Promise<http::Response> updatedProfileMapping;
    EXPECT_CALL(*server.get()->process, profiles(_))
      .WillOnce(Return(http::OK("{}")))
      .WillOnce(Return(updatedProfileMapping.future()));
  
    const Duration pollInterval = Seconds(10);
    loadUriDiskProfileAdaptorModule(
        stringify(server.get()->process->url()),
        pollInterval);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration and prevent retry.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
  
    Clock::pause();
  
    // Register a framework to receive offers.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have
    // wanted resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    Future<vector<Offer>> offersBeforeProfileFound;
    Future<vector<Offer>> offersAfterProfileFound;
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillOnce(DoAll(
          FutureArg<1>(&offersBeforeProfileFound),
          DeclineOffers(declineFilters)))
      .WillOnce(FutureArg<1>(&offersAfterProfileFound));
  
    driver.start();
  
    AWAIT_READY(offersBeforeProfileFound);
    ASSERT_FALSE(offersBeforeProfileFound->empty());
  
    // The offer should not have any resource from the resource provider before
    // the profile appears.
    Resources resourceProviderResources =
      Resources(offersBeforeProfileFound->at(0).resources())
      .filter(&Resources::hasResourceProvider);
  
    EXPECT_TRUE(resourceProviderResources.empty());
  
    // Trigger another poll for profiles. The framework will not receive an offer
    // because there is no change in resources yet.
    Clock::advance(pollInterval);
    Clock::settle();
  
    // Update the disk profile mapping.
    updatedProfileMapping.set(
        http::OK(createDiskProfileMapping({{"test", None()}})));
  
    // Advance the clock to make sure another allocation is triggered.
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offersAfterProfileFound);
    ASSERT_FALSE(offersAfterProfileFound->empty());
  
    // A new storage pool with profile "test" should show up now.
    Resources storagePools =
      Resources(offersAfterProfileFound->at(0).resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"));
  
    EXPECT_FALSE(storagePools.empty());
  }
  
  
  // This test verifies that the storage local resource provider can create then
  // destroy a MOUNT disk from a storage pool with other pipelined operations.
  TEST_P(StorageLocalResourceProviderTest, CreateDestroyDisk)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage/role");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk with a pipelined `RESERVE` operation.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    Resource reserved = *Resources(raw)
      .pushReservation(createDynamicReservationInfo(
          framework.roles(0), framework.principal()))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {RESERVE(reserved),
         CREATE_DISK(reserved, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    ASSERT_TRUE(created.disk().source().has_mount());
    ASSERT_TRUE(created.disk().source().mount().has_root());
    EXPECT_FALSE(path::absolute(created.disk().source().mount().root()));
  
    // Check if the volume is actually created by the test CSI plugin.
    const string& volumePath = created.disk().source().id();
    EXPECT_TRUE(os::exists(volumePath));
  
    // Destroy the MOUNT disk with pipelined `CREATE`, `DESTROY` and `UNRESERVE`
    // operations. Note that `UNRESERVE` can come before `DESTROY_DISK`.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Resource unreserved = *Resources(created)
      .popReservation()
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         DESTROY(persistentVolume),
         UNRESERVE(created),
         DESTROY_DISK(unreserved)});
  
    AWAIT_READY(offers);
  
    // Check if the volume is actually deleted by the test CSI plugin.
    EXPECT_FALSE(os::exists(volumePath));
  }
  
  
  // This test verifies that the storage local resource provider can destroy a
  // MOUNT disk created from a storage pool with other pipelined operations after
  // recovery.
  TEST_P(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage/role");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk with a pipelined `RESERVE` operation.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    Resource reserved = *Resources(raw)
      .pushReservation(createDynamicReservationInfo(
          framework.roles(0), framework.principal()))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {RESERVE(reserved),
         CREATE_DISK(reserved, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    ASSERT_TRUE(created.disk().source().has_mount());
    ASSERT_TRUE(created.disk().source().mount().has_root());
    EXPECT_FALSE(path::absolute(created.disk().source().mount().root()));
  
    // Check if the volume is actually created by the test CSI plugin.
    const string& volumePath = created.disk().source().id();
    EXPECT_TRUE(os::exists(volumePath));
  
    // Restart the agent.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave.get()->terminate();
  
    // NOTE: We set up the resource provider with an extra storage pool, so that
    // when the storage pool shows up, we know that the resource provider has
    // finished reconciling storage pools and thus operations won't be dropped.
    //
    // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
    // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
    // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
    // resource provider reregistration) and wait for the third one, which should
    // contain the extra storage pool. We let it fall through to trigger an offer
    // allocation for the persistent volume.
    //
    // Since the extra storage pool is never used, we reject the offers if only
    // the storage pool is presented.
    //
    // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
    setupResourceProviderConfig(Gigabytes(5));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // NOTE: The order of these expectations is reversed because Google Mock will
    // search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlaveMessage =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
    FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(created)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(updateSlaveMessage);
    ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
    ASSERT_FALSE(Resources(
        updateSlaveMessage->resource_providers().providers(0).total_resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .empty());
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the MOUNT disk with pipelined `CREATE`, `DESTROY` and `UNRESERVE`
    // operations. Note that `UNRESERVE` can come before `DESTROY_DISK`.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Resource unreserved = *Resources(created)
      .popReservation()
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         DESTROY(persistentVolume),
         UNRESERVE(created),
         DESTROY_DISK(unreserved)});
  
    AWAIT_READY(offers);
  
    // Check if the volume is actually deleted by the test CSI plugin.
    EXPECT_FALSE(os::exists(volumePath));
  }
  
  
  // This test verifies that the storage local resource provider can properly
  // handle changes in volume metadata. This is a regression test for MESOS-9395.
  TEST_P(StorageLocalResourceProviderTest, RecoverDiskWithChangedMetadata)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    // Add metadata "label=foo" to each created volume.
    setupResourceProviderConfig(
        Gigabytes(4), None(), None(), None(), "label=foo");
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage/role");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        &Resources::hasResourceProvider)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    RepeatedPtrField<Resource> raw =
      Resources(offer.resources()).filter(&Resources::hasResourceProvider);
  
    ASSERT_EQ(1, raw.size());
    ASSERT_TRUE(isStoragePool(raw[0], "test"));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        &Resources::hasResourceProvider)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE_DISK(raw[0], Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    RepeatedPtrField<Resource> created =
      Resources(offer.resources()).filter(&Resources::hasResourceProvider);
  
    ASSERT_EQ(1, created.size());
    ASSERT_TRUE(isMountDisk(created[0], "test"));
    EXPECT_TRUE(created[0].disk().source().has_metadata());
  
    // Restart the agent.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave.get()->terminate();
  
    // Add metadata "label=bar" to each created volume. This dose not conform to
    // the CSI specification in terms of backward compatibility, but Mesos should
    // be robust to handle changes in the metadata.
    setupResourceProviderConfig(
        Gigabytes(4), None(), None(), None(), "label=bar");
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        Resources::hasResourceProvider)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    RepeatedPtrField<Resource> recovered =
      Resources(offer.resources()).filter(&Resources::hasResourceProvider);
  
    ASSERT_EQ(1, recovered.size()) << "To many resources: " << recovered;
    ASSERT_TRUE(isMountDisk(recovered[0], "test"));
    ASSERT_TRUE(recovered[0].disk().source().has_metadata());
    EXPECT_NE(created[0].disk().source().metadata(),
              recovered[0].disk().source().metadata())
      << "'" << JSON::protobuf(created[0].disk().source().metadata()) << "' vs "
         "'" << JSON::protobuf(recovered[0].disk().source().metadata()) << "'";
  }
  
  
  // This test verifies that the storage local resource provider can properly
  // handle duplicated volumes. This is a regression test for MESOS-9965.
  TEST_P(StorageLocalResourceProviderTest, RecoverDuplicatedVolumes)
  {
    const string mockCsiEndpoint =
      "unix://" + path::join(sandbox.get(), "mock_csi.sock");
  
    MockCSIPlugin plugin;
    ASSERT_SOME(plugin.startup(mockCsiEndpoint));
  
    setupResourceProviderConfig(Bytes(0), None(), mockCsiEndpoint);
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    if (GetParam() == csi::v0::API_VERSION) {
      EXPECT_CALL(plugin, ListVolumes(_, _, A<csi::v0::ListVolumesResponse*>()))
        .WillRepeatedly(Invoke([&](
            grpc::ServerContext* context,
            const csi::v0::ListVolumesRequest* request,
            csi::v0::ListVolumesResponse* response) {
          csi::v0::Volume volume;
          volume.set_capacity_bytes(Gigabytes(2).bytes());
          volume.set_id("volume1");
  
          // Report duplicated volumes.
          *response->add_entries()->mutable_volume() = volume;
          *response->add_entries()->mutable_volume() = volume;
  
          return grpc::Status::OK;
        }));
    } else {
      EXPECT_CALL(plugin, ListVolumes(_, _, A<csi::v1::ListVolumesResponse*>()))
        .WillRepeatedly(Invoke([&](
            grpc::ServerContext* context,
            const csi::v1::ListVolumesRequest* request,
            csi::v1::ListVolumesResponse* response) {
          csi::v1::Volume volume;
          volume.set_capacity_bytes(Gigabytes(2).bytes());
          volume.set_volume_id("volume1");
  
          // Report duplicated volumes.
          *response->add_entries()->mutable_volume() = volume;
          *response->add_entries()->mutable_volume() = volume;
  
          return grpc::Status::OK;
        }));
    }
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage/role");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        &Resources::hasResourceProvider)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
  
    // Restart the agent.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave.get()->terminate();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        &Resources::hasResourceProvider)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Wait for an offer to verify that the resource provider comes back.
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    Resources resourceProviderResources =
      Resources(offer.resources()).filter(&Resources::hasResourceProvider);
  
    EXPECT_SOME_EQ(Gigabytes(2), resourceProviderResources.disk());
  }
  
  
  // This test verifies that a framework cannot create a volume during and after
  // the profile disappears, and destroying a volume with a stale profile will
  // recover the freed disk with another appeared profile.
  //
  // To accomplish this:
  //   1. Create a 2GB MOUNT disk from a 4GB RAW disk of profile 'test1'.
  //   2. Create another MOUNT disk from the rest RAW disk of profile 'test1'.
  //   3. Remove profile 'test1' and adds profile 'test2' before the second
  //      operation is applied. The operation would then be dropped, and the rest
  //      RAW disk would show up as of profile 'test2'.
  //   4. Destroy the MOUNT disk of profile 'test1'. All 4GB RAW disk would show
  //      up as of profile 'test2'.
  TEST_P(StorageLocalResourceProviderTest, ProfileDisappeared)
  {
    Clock::pause();
  
    Future<shared_ptr<TestDiskProfileServer>> server =
      TestDiskProfileServer::create();
  
    AWAIT_READY(server);
  
    Promise<http::Response> updatedProfileMapping;
    EXPECT_CALL(*server.get()->process, profiles(_))
      .WillOnce(Return(http::OK(createDiskProfileMapping({{"test1", None()}}))))
      .WillOnce(Return(updatedProfileMapping.future()))
      .WillRepeatedly(Return(Future<http::Response>())); // Stop subsequent polls.
  
    const Duration pollInterval = Seconds(10);
    loadUriDiskProfileAdaptorModule(
        stringify(server.get()->process->url()),
        pollInterval);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration and prevent retry.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
  
    Clock::pause();
  
    // Register a v1 framework to exercise operations with feedback.
    v1::FrameworkInfo framework = v1::DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
  
    EXPECT_CALL(*scheduler, connected(_))
      .WillOnce(v1::scheduler::SendSubscribe(framework));
  
    Future<v1::scheduler::Event::Subscribed> subscribed;
    EXPECT_CALL(*scheduler, subscribed(_, _))
      .WillOnce(FutureArg<1>(&subscribed));
  
    EXPECT_CALL(*scheduler, heartbeat(_))
      .WillRepeatedly(Return()); // Ignore heartbeats.
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(*scheduler, offers(_, _))
      .WillRepeatedly(v1::scheduler::DeclineOffers());
  
    Future<v1::scheduler::Event::Offers> offers;
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(
            std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"))))
      .WillOnce(FutureArg<1>(&offers));
  
    v1::scheduler::TestMesos mesos(
        master.get()->pid, ContentType::PROTOBUF, scheduler);
  
    AWAIT_READY(subscribed);
  
    const v1::FrameworkID& frameworkId = subscribed->framework_id();
  
    // NOTE: If the framework has not declined an unwanted offer yet when the
    // resource provider reports its RAW resources, the new allocation triggered
    // by this update won't generate an allocatable offer due to no CPU and memory
    // resources. So we first settle the clock to ensure that the unwanted offer
    // has been declined, then advance the clock to trigger another allocation.
    Clock::settle();
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1, offers->offers_size());
  
    v1::Offer offer = offers->offers(0);
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(
            std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"))))
      .WillOnce(FutureArg<1>(&offers));
  
    // Create a 2GB MOUNT disk of profile 'test1'.
    {
      v1::Resources raw = v1::Resources(offer.resources())
        .filter(std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"));
  
      ASSERT_SOME_EQ(Gigabytes(4), raw.disk());
  
      // Just use 2GB of the storage pool.
      v1::Resource source = *raw.begin();
      source.mutable_scalar()->set_value(
          static_cast<double>(Gigabytes(2).bytes()) / Bytes::MEGABYTES);
  
      v1::OperationID operationId;
      operationId.set_value(id::UUID::random().toString());
  
      Future<v1::scheduler::Event::UpdateOperationStatus> update;
      EXPECT_CALL(
          *scheduler,
          updateOperationStatus(
              _, v1::scheduler::OperationStatusUpdateOperationIdEq(operationId)))
        .WillOnce(FutureArg<1>(&update))
        .WillRepeatedly(Return()); // Ignore subsequent updates.
  
      mesos.send(v1::createCallAccept(
          frameworkId,
          offer,
          {v1::CREATE_DISK(
               source,
               v1::Resource::DiskInfo::Source::MOUNT,
               None(),
               operationId)}));
  
      AWAIT_READY(update);
      EXPECT_EQ(v1::OPERATION_FINISHED, update->status().state());
    }
  
    // Advance the clock to trigger another allocation.
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1, offers->offers_size());
  
    offer = offers->offers(0);
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(
            std::bind(isMountDisk<v1::Resource>, lambda::_1, "test1"))))
      .WillOnce(FutureArg<1>(&offers));
  
    // We drop the agent update (which is triggered by the changes in the known
    // set of profiles) to simulate the situation where the update races with
    // an offer operation.
    Future<UpdateSlaveMessage> updateSlave3 =
      DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    // Remove profile 'test1' and add profile 'test2'. No allocation will be
    // triggered since the framework is still holding the current offer.
    Clock::advance(pollInterval);
    updatedProfileMapping.set(
        http::OK(createDiskProfileMapping({{"test2", None()}})));
  
    AWAIT_READY(updateSlave3);
  
    // Create another MOUNT disk from the rest RAW disk of profile 'test1'. This
    // operation would be dropped due to a mismatched resource version.
    {
      v1::Resources raw = v1::Resources(offer.resources())
        .filter(std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"));
  
      ASSERT_SOME_EQ(Gigabytes(2), raw.disk());
  
      v1::OperationID operationId;
      operationId.set_value(id::UUID::random().toString());
  
      Future<v1::scheduler::Event::UpdateOperationStatus> update;
      EXPECT_CALL(
          *scheduler,
          updateOperationStatus(
              _, v1::scheduler::OperationStatusUpdateOperationIdEq(operationId)))
        .WillOnce(FutureArg<1>(&update))
        .WillRepeatedly(Return()); // Ignore subsequent updates;
  
      mesos.send(v1::createCallAccept(
          frameworkId,
          offer,
          {v1::CREATE_DISK(
               *raw.begin(),
               v1::Resource::DiskInfo::Source::MOUNT,
               None(),
               operationId)}));
  
      AWAIT_READY(update);
      EXPECT_EQ(v1::OPERATION_DROPPED, update->status().state());
    }
  
    // Forward the dropped agent update to trigger another allocation.
    post(slave.get()->pid, master.get()->pid, updateSlave3.get());
  
    AWAIT_READY(offers);
    ASSERT_EQ(1, offers->offers_size());
  
    offer = offers->offers(0);
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(
            std::bind(isStoragePool<v1::Resource>, lambda::_1, "test2"))))
      .WillOnce(FutureArg<1>(&offers));
  
    // Destroy the MOUNT disk of profile 'test1'. The returned converted resources
    // should be empty.
    {
      v1::Resources created = v1::Resources(offer.resources())
        .filter(std::bind(isMountDisk<v1::Resource>, lambda::_1, "test1"));
  
      ASSERT_SOME_EQ(Gigabytes(2), created.disk());
  
      v1::OperationID operationId;
      operationId.set_value(id::UUID::random().toString());
  
      Future<v1::scheduler::Event::UpdateOperationStatus> update;
      EXPECT_CALL(
          *scheduler,
          updateOperationStatus(
              _, v1::scheduler::OperationStatusUpdateOperationIdEq(operationId)))
        .WillOnce(FutureArg<1>(&update))
        .WillRepeatedly(Return()); // Ignore subsequent updates;
  
      mesos.send(v1::createCallAccept(
          frameworkId, offer, {v1::DESTROY_DISK(*created.begin(), operationId)}));
  
      AWAIT_READY(update);
      EXPECT_EQ(v1::OPERATION_FINISHED, update->status().state());
      EXPECT_TRUE(update->status().converted_resources().empty());
    }
  
    // The resource provider will reconcile the storage pools to reclaim the
    // space freed by destroying a MOUNT disk of a disappeared profile, which
    // would in turn trigger another agent update and thus another allocation.
    AWAIT_READY(offers);
    ASSERT_EQ(1, offers->offers_size());
  
    offer = offers->offers(0);
  
    // Check that the freed disk shows up as of profile 'test2'.
    {
      v1::Resources raw = v1::Resources(offer.resources())
        .filter(std::bind(isStoragePool<v1::Resource>, lambda::_1, "test2"));
  
      EXPECT_SOME_EQ(Gigabytes(4), raw.disk());
    }
  }
  
  
  // This test verifies that the storage local resource provider can
  // recover if the plugin is killed during an agent failover..
  TEST_P(StorageLocalResourceProviderTest, AgentFailoverPluginKilled)
  {
    setupResourceProviderConfig(Bytes(0), "volume0:4GB");
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
  
    slave::Fetcher fetcher(slaveFlags);
  
    Try<slave::MesosContainerizer*> _containerizer =
      slave::MesosContainerizer::create(slaveFlags, false, &fetcher);
    ASSERT_SOME(_containerizer);
  
    Owned<slave::MesosContainerizer> containerizer(_containerizer.get());
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave =
      StartSlave(detector.get(), containerizer.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to receive offers.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // The framework is expected to see the following offers in sequence:
    //   1. One containing any resource from the resource provider before the
    //      agent fails over.
    //   2. One containing any resource from the resource provider after the agent
    //      recovers from the failover.
    //
    // We set up the expectations for these offers as the test progresses.
    Future<vector<Offer>> rawDiskOffers;
    Future<vector<Offer>> slaveRecoveredOffers;
  
    // We use the following filter to filter offers that do not have
    // wanted resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline offers that contain only the agent's default resources.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        &Resources::hasResourceProvider)))
      .WillOnce(FutureArg<1>(&rawDiskOffers));
  
    driver.start();
  
    AWAIT_READY(rawDiskOffers);
  
    // Get the PID of the plugin container so we can kill it later.
    Future<hashset<ContainerID>> pluginContainers = containerizer->containers();
  
    AWAIT_READY(pluginContainers);
    ASSERT_EQ(1u, pluginContainers->size());
  
    Future<ContainerStatus> pluginStatus =
      containerizer->status(*pluginContainers->begin());
  
    AWAIT_READY(pluginStatus);
  
    // Terminate the agent to simulate a failover.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave.get()->terminate();
    slave->reset();
    containerizer.reset();
  
    // Kill the plugin container.
    Future<Option<int>> pluginReaped = reap(pluginStatus->executor_pid());
    ASSERT_EQ(0, os::kill(pluginStatus->executor_pid(), SIGKILL));
  
    AWAIT_READY(pluginReaped);
    ASSERT_SOME(pluginReaped.get());
    ASSERT_TRUE(WIFSIGNALED(pluginReaped->get()));
    EXPECT_EQ(SIGKILL, WTERMSIG(pluginReaped->get()));
  
    // Restart the agent.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        &Resources::hasResourceProvider)))
      .WillOnce(FutureArg<1>(&slaveRecoveredOffers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRecoveredOffers);
  }
  
  
  // This test verifies that if an agent is registered with a new ID, the resource
  // provider ID would change as well, and any created volume becomes a
  // preprovisioned volume. A framework should be able to either create a MOUNT
  // disk from a preprovisioned volume and use it, or destroy it directly.
  TEST_P(StorageLocalResourceProviderTest, ROOT_AgentRegisteredWithNewId)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    // We set up a 'good' and a 'bad' profile to test that `CREATE_DISK` only
    // succeeds with the right profile. Ideally this should be tested in
    // `CreateDestroyPreprovisionedVolume`, but since CSI v0 doesn't support
    // validating a volume against parameters, we have to test it here.
    ASSERT_SOME(os::write(profilesPath, createDiskProfileMapping(
        {{"good", JSON::Object{{"label", "foo"}}},
         {"bad", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(5), None(), None(), "label=foo");
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers, e.g., offers containing only agent-default
    // resources.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // We first wait for an offer containing a storage pool to exercise two
    // `CREATE_DISK` operations. Since they are not done atomically, we might get
    // subsequent offers containing a smaller storage pool after one is exercised,
    // so we decline them.
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "good"))))
      .WillOnce(FutureArg<1>(&offers))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    Resources raw = Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "good"));
  
    // Create a 2GB and a 3GB MOUNT disks.
    ASSERT_SOME_EQ(Gigabytes(5), raw.disk());
    Resource source1 = *raw.begin();
    source1.mutable_scalar()->set_value(
        static_cast<double>(Gigabytes(2).bytes()) / Bytes::MEGABYTES);
  
    raw -= source1;
    ASSERT_SOME_EQ(Gigabytes(3), raw.disk());
    Resource source2 = *raw.begin();
  
    // After the following operations are completed, we would get an offer
    // containing __two__ MOUNT disks: one 2GB and one 3GB.
    EXPECT_CALL(sched, resourceOffers(&driver, AllOf(
        OffersHaveAnyResource([](const Resource& r) {
          return isMountDisk(r, "good") &&
            Megabytes(r.scalar().value()) == Gigabytes(2);
        }),
        OffersHaveAnyResource([](const Resource& r) {
          return isMountDisk(r, "good") &&
            Megabytes(r.scalar().value()) == Gigabytes(3);
        }))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE_DISK(source1, Resource::DiskInfo::Source::MOUNT),
         CREATE_DISK(source2, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    std::vector<Resource> created = google::protobuf::convert<Resource>(
        Resources(offer.resources())
          .filter(std::bind(isMountDisk<Resource>, lambda::_1, "good")));
  
    ASSERT_EQ(2u, created.size());
  
    // Create persistent MOUNT volumes then launch a tasks to write files.
    Resources persistentVolumes;
    hashmap<string, ResourceProviderID> sourceIdToProviderId;
    const vector<string> containerPaths = {"volume1", "volume2"};
    for (size_t i = 0; i < created.size(); i++) {
      const Resource& volume = created[i];
      ASSERT_TRUE(volume.has_provider_id());
      sourceIdToProviderId.put(volume.disk().source().id(), volume.provider_id());
  
      Resource persistentVolume = volume;
      persistentVolume.mutable_disk()->mutable_persistence()
        ->set_id(id::UUID::random().toString());
      persistentVolume.mutable_disk()->mutable_persistence()
        ->set_principal(framework.principal());
      persistentVolume.mutable_disk()->mutable_volume()
        ->set_container_path(containerPaths[i]);
      persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
      persistentVolumes += persistentVolume;
    }
  
    Future<Nothing> taskFinished;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    // After the task is finished, we would get an offer containing the two
    // persistent MOUNT volumes. We just match any one here for simplicity.
    EXPECT_CALL( sched, resourceOffers(&driver, OffersHaveAnyResource(
        Resources::isPersistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolumes),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolumes),
             createCommandInfo(
                 "touch " + path::join(containerPaths[0], "file") + " " +
                 path::join(containerPaths[1], "file")))})});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(offers);
  
    // Shut down the agent.
    EXPECT_CALL(sched, offerRescinded(&driver, _));
  
    slave.get()->terminate();
  
    // Remove the `latest` symlink to register the agent with a new ID.
    const string metaDir = slave::paths::getMetaRootDir(slaveFlags.work_dir);
    ASSERT_SOME(os::rm(slave::paths::getLatestSlavePath(metaDir)));
  
    // NOTE: We set up the resource provider with an extra storage pool, so that
    // when the storage pool shows up, we know that the resource provider has
    // finished reconciling storage pools and thus operations won't be dropped.
    //
    // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
    // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
    // `UpdateSlaveMessage`s (one sent after agent registration and one after
    // resource provider registration) and wait for the third one, which should
    // contain the extra storage pool. We let it fall through to trigger an offer
    // allocation for the persistent volume.
    //
    // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
    setupResourceProviderConfig(Gigabytes(6), None(), None(), "label=foo");
  
    // NOTE: The order of these expectations is reversed because Google Mock will
    // search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlaveMessage =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUFS(SlaveRegisteredMessage(), _, _);
  
    Future<SlaveRegisteredMessage> newSlaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, Not(slave.get()->pid));
  
    // After the new agent starts, we wait for an offer containing the two
    // preprovisioned volumes to exercise two `CREATE_DISK` operations. Since one
    // of them would fail, we might get subsequent offers containing a
    // preprovisioned volumes afterwards, so we decline them.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        isPreprovisionedVolume<Resource>)))
      .WillOnce(FutureArg<1>(&offers))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(newSlaveRegisteredMessage);
    ASSERT_NE(slaveRegisteredMessage->slave_id(),
              newSlaveRegisteredMessage->slave_id());
  
    AWAIT_READY(updateSlaveMessage);
    ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
    ASSERT_FALSE(Resources(
        updateSlaveMessage->resource_providers().providers(0).total_resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "good"))
      .empty());
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    vector<Resource> preprovisioned = google::protobuf::convert<Resource>(
        Resources(offer.resources()).filter(isPreprovisionedVolume<Resource>));
  
    ASSERT_EQ(2u, preprovisioned.size());
  
    // Check that the resource provider IDs of the volumes have changed.
    vector<string> volumePaths;
    foreach (const Resource& volume, preprovisioned) {
      ASSERT_TRUE(volume.has_provider_id());
      ASSERT_TRUE(sourceIdToProviderId.contains(volume.disk().source().id()));
      ASSERT_NE(sourceIdToProviderId.at(volume.disk().source().id()),
                volume.provider_id());
  
      const string& volumePath = volume.disk().source().id();
      ASSERT_TRUE(os::exists(volumePath));
      volumePaths.push_back(volumePath);
    }
  
    // Apply profiles 'good' and 'bad' to the preprovisioned volumes. The first
    // operation would succeed but the second would fail.
    Future<multiset<OperationState>> createDiskOperationStates =
      process::collect(
          FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _),
          FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _))
        .then([](const std::tuple<
                  UpdateOperationStatusMessage,
                  UpdateOperationStatusMessage>& operationStatuses) {
          return multiset<OperationState>{
            std::get<0>(operationStatuses).status().state(),
            std::get<1>(operationStatuses).status().state()};
        });
  
    // After both operations are completed, we would get an offer containing both
    // a MOUNT disk and a preprovisioned volume.
    EXPECT_CALL(sched, resourceOffers(&driver, AllOf(
        OffersHaveAnyResource(
            std::bind(isMountDisk<Resource>, lambda::_1, "good")),
        OffersHaveResource(preprovisioned[1]))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE_DISK(
             preprovisioned[0], Resource::DiskInfo::Source::MOUNT, "good"),
         CREATE_DISK(
             preprovisioned[1], Resource::DiskInfo::Source::MOUNT, "bad")});
  
    multiset<OperationState> expectedOperationStates = {
      OperationState::OPERATION_FINISHED,
      OperationState::OPERATION_FAILED};
  
    AWAIT_EXPECT_EQ(expectedOperationStates, createDiskOperationStates);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource imported = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "good"))
      .begin();
  
    // Create persistent MOUNT volumes from the imported volume then launch a
    // tasks to write files, and destroy the other unimported volume.
    imported.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    imported.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    imported.mutable_disk()->mutable_volume()->set_container_path("volume");
    imported.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    Future<UpdateOperationStatusMessage> destroyDiskOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    // After the task is finished and the operation is completed, we would get an
    // offer the persistent volume and a storage pool that has the freed space.
    EXPECT_CALL(sched, resourceOffers(&driver, AllOf(
        OffersHaveResource(imported),
        OffersHaveAnyResource([](const Resource& r) {
          return isStoragePool(r, "good") &&
            Megabytes(r.scalar().value()) >= Gigabytes(2);
        }))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(imported),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(imported),
             createCommandInfo("test -f " + path::join("volume", "file")))}),
         DESTROY_DISK(preprovisioned[1])});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(destroyDiskOperationStatus);
    EXPECT_EQ(OperationState::OPERATION_FINISHED,
              destroyDiskOperationStatus->status().state());
  
    AWAIT_READY(offers);
  
    // Check if the unimported volume is deleted by the test CSI plugin.
    EXPECT_FALSE(os::exists(volumePaths[1]));
  }
  
  
  // This test verifies that the storage local resource provider can create and
  // publish a persistent volume used by a task, and the persistent volume will be
  // cleaned up when being destroyed.
  //
  // To accomplish this:
  //   1. Create a MOUNT disk from a RAW disk resource.
  //   2. Create a persistent volume on the MOUNT disk then launches a task to
  //      write a file into it.
  //   3. Destroy the persistent volume but keep the MOUNT disk. The file should
  //      be deleted.
  //   4. Destroy the MOUNT disk.
  TEST_P(
      StorageLocalResourceProviderTest, ROOT_CreateDestroyPersistentMountVolume)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    ASSERT_TRUE(created.disk().source().has_mount());
    ASSERT_TRUE(created.disk().source().mount().has_root());
    EXPECT_FALSE(path::absolute(created.disk().source().mount().root()));
  
    // Check if the CSI volume is actually created.
    const string& volumePath = created.disk().source().id();
    EXPECT_TRUE(os::exists(volumePath));
  
    // Create a persistent MOUNT volume then launch a task to write a file.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<Nothing> taskFinished;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("touch " + path::join("volume", "file")))})});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the persistent volume.
    Future<UpdateOperationStatusMessage> updateOperationStatusMessage =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(created)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers({offer.id()}, {DESTROY(persistentVolume)});
  
    // NOTE: Since `DESTROY` would be applied by the master synchronously, we
    // might get an offer before the persistent volume is cleaned up on the agent,
    // so we wait for an `UpdateOperationStatusMessage` before the check below.
    AWAIT_READY(updateOperationStatusMessage);
    EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
  
    // Check if the CSI volume still exists but has being cleaned up.
    EXPECT_TRUE(os::exists(volumePath));
    EXPECT_FALSE(os::exists(path::join(volumePath, "file")));
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the MOUNT disk.
    updateOperationStatusMessage =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers({offer.id()}, {DESTROY_DISK(created)});
  
    AWAIT_READY(updateOperationStatusMessage);
    EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
  
    // Check if the CSI volume is actually deleted.
    EXPECT_FALSE(os::exists(volumePath));
  
    AWAIT_READY(offers);
  }
  
  
  // This test verifies that the storage local resource provider can republish and
  // clean up persistent volumes after recovery.
  //
  // To accomplish this:
  //   1. Create a MOUNT disk from a RAW disk resources.
  //   2. Create a persistent volume on the MOUNT disk then launch a task to
  //      write a file into it.
  //   3. Restart the agent.
  //   4. Launch another task to read the file.
  //   5. Restart the agent again.
  //   6. Destroy the persistent volume but keep the MOUNT disk. The file should
  //      be deleted.
  //   7. Destroy the MOUNT disk.
  TEST_P(
      StorageLocalResourceProviderTest,
      ROOT_CreateDestroyPersistentMountVolumeWithRecovery)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Create a persistent MOUNT volume then launch a task to write a file.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<Nothing> taskFinished;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("touch " + path::join("volume", "file")))})});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(offers);
  
    // Restart the agent.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave.get()->terminate();
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Launch another task to read the file created by the previous task.
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("test -f " + path::join("volume", "file")))})});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(offers);
  
    // Restart the agent.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave.get()->terminate();
  
    // NOTE: We set up the resource provider with an extra storage pool, so that
    // when the storage pool shows up, we know that the resource provider has
    // finished reconciling storage pools and thus operations won't be dropped.
    //
    // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
    // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
    // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
    // resource provider reregistration) and wait for the third one, which should
    // contain the extra storage pool. We let it fall through to trigger an offer
    // allocation for the persistent volume.
    //
    // Since the extra storage pool is never used, we reject the offers if only
    // the storage pool is presented.
    //
    // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
    setupResourceProviderConfig(Gigabytes(5));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // NOTE: The order of these expectations is reversed because Google Mock will
    // search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlaveMessage =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
    FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(updateSlaveMessage);
    ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
    ASSERT_FALSE(Resources(
        updateSlaveMessage->resource_providers().providers(0).total_resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .empty());
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the persistent volume.
    Future<UpdateOperationStatusMessage> updateOperationStatusMessage =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(created)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers({offer.id()}, {DESTROY(persistentVolume)});
  
    // NOTE: Since `DESTROY` would be applied by the master synchronously, we
    // might get an offer before the persistent volume is cleaned up on the agent,
    // so we wait for an `UpdateOperationStatusMessage` before the check below.
    AWAIT_READY(updateOperationStatusMessage);
    EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
  
    // Check if the CSI volume still exists but has being cleaned up.
    const string& volumePath = created.disk().source().id();
    EXPECT_TRUE(os::exists(volumePath));
    EXPECT_FALSE(os::exists(path::join(volumePath, "file")));
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the MOUNT disk.
    updateOperationStatusMessage =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers({offer.id()}, {DESTROY_DISK(created)});
  
    AWAIT_READY(updateOperationStatusMessage);
    EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
  
    // Check if the CSI volume is actually deleted.
    EXPECT_FALSE(os::exists(volumePath));
  
    AWAIT_READY(offers);
  }
  
  
  // This test verifies that the storage local resource provider can republish and
  // clean up persistent volumes after agent reboot.
  //
  // To accomplish this:
  //   1. Create a MOUNT disk from a RAW disk resource.
  //   2. Create a persistent volume on the MOUNT disk then launches a task to
  //      write a file into it.
  //   3. Simulate an agent reboot.
  //   4. Launch another task to read the file.
  //   5. Simulate another agent reboot.
  //   6. Destroy the persistent volume but keep the MOUNT disk. The file should
  //      be deleted.
  //   7. Destroy the MOUNT disk.
  TEST_P(
      StorageLocalResourceProviderTest,
      ROOT_CreateDestroyPersistentMountVolumeWithReboot)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Create a persistent MOUNT volume then launch a task to write a file.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<Nothing> taskFinished;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("touch " + path::join("volume", "file")))})});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(offers);
  
    // Shutdown the agent and unmount all CSI volumes to simulate the 1st reboot.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave->reset();
  
    const string csiRootDir = slave::paths::getCsiRootDir(slaveFlags.work_dir);
    ASSERT_SOME(fs::unmountAll(csiRootDir));
  
    // Inject the boot IDs to simulate the 1st reboot.
    auto injectBootIds = [&](const string& bootId) {
      ASSERT_SOME(os::write(
          slave::paths::getBootIdPath(
              slave::paths::getMetaRootDir(slaveFlags.work_dir)),
          bootId));
  
      Try<list<string>> volumePaths =
        csi::paths::getVolumePaths(csiRootDir, "*", "*");
      ASSERT_SOME(volumePaths);
      ASSERT_FALSE(volumePaths->empty());
  
      foreach (const string& path, volumePaths.get()) {
        Try<csi::paths::VolumePath> volumePath =
          csi::paths::parseVolumePath(csiRootDir, path);
        ASSERT_SOME(volumePath);
  
        const string volumeStatePath = csi::paths::getVolumeStatePath(
            csiRootDir, volumePath->type, volumePath->name, volumePath->volumeId);
  
        Result<csi::state::VolumeState> volumeState =
          slave::state::read<csi::state::VolumeState>(volumeStatePath);
  
        ASSERT_SOME(volumeState);
  
        if (volumeState->state() == csi::state::VolumeState::PUBLISHED) {
          volumeState->set_boot_id(bootId);
          ASSERT_SOME(
              slave::state::checkpoint(volumeStatePath, volumeState.get()));
        }
      }
    };
  
    injectBootIds("1st reboot! ;)");
  
    // Restart the agent.
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Launch another task to read the file created by the previous task.
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("test -f " + path::join("volume", "file")))})});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(offers);
  
    // Shutdown the agent and unmount all CSI volumes to simulate the 2nd reboot.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave->reset();
    ASSERT_SOME(fs::unmountAll(csiRootDir));
  
    // Inject the boot IDs to simulate the 2nd reboot.
    injectBootIds("2nd reboot! ;)");
  
    // NOTE: We set up the resource provider with an extra storage pool, so that
    // when the storage pool shows up, we know that the resource provider has
    // finished reconciling storage pools and thus operations won't be dropped.
    //
    // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
    // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
    // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
    // resource provider reregistration) and wait for the third one, which should
    // contain the extra storage pool. We let it fall through to trigger an offer
    // allocation for the persistent volume.
    //
    // Since the extra storage pool is never used, we reject the offers if only
    // the storage pool is presented.
    //
    // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
    setupResourceProviderConfig(Gigabytes(5));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // NOTE: The order of these expectations is reversed because Google Mock will
    // search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlaveMessage =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
    FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
  
    // Restart the agent.
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(updateSlaveMessage);
    ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
    ASSERT_FALSE(Resources(
        updateSlaveMessage->resource_providers().providers(0).total_resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .empty());
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the persistent volume.
    Future<UpdateOperationStatusMessage> updateOperationStatusMessage =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(created)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers({offer.id()}, {DESTROY(persistentVolume)});
  
    // NOTE: Since `DESTROY` would be applied by the master synchronously, we
    // might get an offer before the persistent volume is cleaned up on the agent,
    // so we wait for an `UpdateOperationStatusMessage` before the check below.
    AWAIT_READY(updateOperationStatusMessage);
    EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
  
    // Check if the CSI volume still exists but has being cleaned up.
    const string& volumePath = created.disk().source().id();
    EXPECT_TRUE(os::exists(volumePath));
    EXPECT_FALSE(os::exists(path::join(volumePath, "file")));
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the MOUNT disk.
    updateOperationStatusMessage =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers({offer.id()}, {DESTROY_DISK(created)});
  
    AWAIT_READY(updateOperationStatusMessage);
    EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
  
    // Check if the CSI volume is actually deleted.
    EXPECT_FALSE(os::exists(volumePath));
  
    AWAIT_READY(offers);
  }
  
  
  // This test verifies that the storage local resource provider can
  // restart its CSI plugin after it is killed and continue to work
  // properly.
  TEST_P(
      StorageLocalResourceProviderTest,
      ROOT_PublishUnpublishResourcesPluginKilled)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.isolation = "filesystem/linux";
  
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    slave::Fetcher fetcher(slaveFlags);
  
    Try<slave::MesosContainerizer*> _containerizer =
      slave::MesosContainerizer::create(slaveFlags, false, &fetcher);
    ASSERT_SOME(_containerizer);
  
    Owned<slave::MesosContainerizer> containerizer(_containerizer.get());
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave =
      StartSlave(detector.get(), containerizer.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // The framework is expected to see the following offers in sequence:
    //   1. One containing a RAW disk resource before `CREATE_DISK`.
    //   2. One containing a MOUNT disk resource after `CREATE_DISK`.
    //   3. One containing the same MOUNT disk resource after `CREATE`,
    //      `LAUNCH` and `DESTROY`.
    //   4. One containing the same RAW disk resource after `DESTROY_DISK`.
    //
    // We set up the expectations for these offers as the test progresses.
    Future<vector<Offer>> rawDiskOffers;
    Future<vector<Offer>> volumeCreatedOffers;
    Future<vector<Offer>> taskFinishedOffers;
    Future<vector<Offer>> volumeDestroyedOffers;
  
    Sequence offers;
  
    // We use the following filter to filter offers that do not have
    // wanted resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline offers that contain only the agent's default resources.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&rawDiskOffers));
  
    driver.start();
  
    AWAIT_READY(rawDiskOffers);
    ASSERT_FALSE(rawDiskOffers->empty());
  
    Option<Resource> source;
  
    foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
        break;
      }
    }
  
    ASSERT_SOME(source);
  
    // Get the ID of the CSI plugin container.
    Future<hashset<ContainerID>> pluginContainers = containerizer->containers();
  
    AWAIT_READY(pluginContainers);
    ASSERT_EQ(1u, pluginContainers->size());
  
    const ContainerID& pluginContainerId = *pluginContainers->begin();
  
    Future<Nothing> pluginRestarted =
      FUTURE_DISPATCH(_, &ContainerDaemonProcess::launchContainer);
  
    // Kill the plugin container and wait for it to restart.
    Future<int> pluginKilled = containerizer->status(pluginContainerId)
      .then([](const ContainerStatus& status) {
        return os::kill(status.executor_pid(), SIGKILL);
      });
  
    AWAIT_ASSERT_EQ(0, pluginKilled);
    AWAIT_READY(pluginRestarted);
  
    // Create a volume.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&volumeCreatedOffers));
  
    driver.acceptOffers(
        {rawDiskOffers->at(0).id()},
        {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(volumeCreatedOffers);
    ASSERT_FALSE(volumeCreatedOffers->empty());
  
    Option<Resource> volume;
  
    foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
      if (isMountDisk(resource, "test")) {
        volume = resource;
        break;
      }
    }
  
    ASSERT_SOME(volume);
    ASSERT_TRUE(volume->disk().source().has_vendor());
    EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
    ASSERT_TRUE(volume->disk().source().has_id());
    ASSERT_TRUE(volume->disk().source().has_mount());
    ASSERT_TRUE(volume->disk().source().mount().has_root());
    EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
  
    // Check if the volume is actually created by the test CSI plugin.
    const string& volumePath = volume->disk().source().id();
    EXPECT_TRUE(os::exists(volumePath));
  
    pluginRestarted =
      FUTURE_DISPATCH(_, &ContainerDaemonProcess::launchContainer);
  
    // Kill the plugin container and wait for it to restart.
    pluginKilled = containerizer->status(pluginContainerId)
      .then([](const ContainerStatus& status) {
        return os::kill(status.executor_pid(), SIGKILL);
      });
  
    AWAIT_ASSERT_EQ(0, pluginKilled);
    AWAIT_READY(pluginRestarted);
  
    // Put a file into the volume.
    ASSERT_SOME(os::touch(path::join(volumePath, "file")));
  
    // Create a persistent volume on the CSI volume, then launch a task to
    // use the persistent volume.
    Resource persistentVolume = volume.get();
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<TaskStatus> taskStarting;
    Future<TaskStatus> taskRunning;
    Future<TaskStatus> taskFinished;
  
    EXPECT_CALL(sched, statusUpdate(&driver, _))
      .WillOnce(FutureArg<1>(&taskStarting))
      .WillOnce(FutureArg<1>(&taskRunning))
      .WillOnce(FutureArg<1>(&taskFinished));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
        persistentVolume)))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&taskFinishedOffers));
  
    driver.acceptOffers(
        {volumeCreatedOffers->at(0).id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             volumeCreatedOffers->at(0).slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("test -f " + path::join("volume", "file")))})});
  
    AWAIT_READY(taskStarting);
    EXPECT_EQ(TASK_STARTING, taskStarting->state());
  
    AWAIT_READY(taskRunning);
    EXPECT_EQ(TASK_RUNNING, taskRunning->state());
  
    AWAIT_READY(taskFinished);
    EXPECT_EQ(TASK_FINISHED, taskFinished->state());
  
    AWAIT_READY(taskFinishedOffers);
  
    pluginRestarted =
      FUTURE_DISPATCH(_, &ContainerDaemonProcess::launchContainer);
  
    // Kill the plugin container and wait for it to restart.
    pluginKilled = containerizer->status(pluginContainerId)
      .then([](const ContainerStatus& status) {
        return os::kill(status.executor_pid(), SIGKILL);
      });
  
    AWAIT_ASSERT_EQ(0, pluginKilled);
    AWAIT_READY(pluginRestarted);
  
    // Destroy the persistent volume and the CSI volume.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(source.get())))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&volumeDestroyedOffers));
  
    driver.acceptOffers(
        {taskFinishedOffers->at(0).id()},
        {DESTROY(persistentVolume), DESTROY_DISK(volume.get())});
  
    AWAIT_READY(volumeDestroyedOffers);
    ASSERT_FALSE(volumeDestroyedOffers->empty());
  
    // Check if the volume is actually deleted by the test CSI plugin.
    EXPECT_FALSE(os::exists(volumePath));
  }
  
  
  // This test verifies that the storage local resource provider would fail to
  // create a persistent volume on a BLOCK disk resource.
  //
  // TODO(chhsiao): Update this test once persistent BLOCK volumes are supported.
  TEST_P(StorageLocalResourceProviderTest, CreatePersistentBlockVolume)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
    Try<string> blockDiskProfileMapping = strings::format(
        R"~(
        {
          "profile_matrix": {
            "test": {
              "csi_plugin_type_selector": {
                "plugin_type": "%s"
              },
              "volume_capabilities": {
                "block": {},
                "access_mode": {
                  "mode": "SINGLE_NODE_WRITER"
                }
              }
            }
          }
        }
        )~",
        TEST_CSI_PLUGIN_TYPE);
  
    ASSERT_SOME(blockDiskProfileMapping);
    ASSERT_SOME(os::write(profilesPath, blockDiskProfileMapping.get()));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    const string mockCsiEndpoint =
      "unix://" + path::join(sandbox.get(), "mock_csi.sock");
  
    // We use a mock CSI plugin here because the test CSI plugin does not support
    // block volumes yet.
    MockCSIPlugin plugin;
    ASSERT_SOME(plugin.startup(mockCsiEndpoint));
  
    setupResourceProviderConfig(Bytes(0), None(), mockCsiEndpoint);
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a BLOCK disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isBlockDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::BLOCK)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isBlockDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Create a persistent BLOCK volume, which would fail.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<UpdateOperationStatusMessage> createOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(created)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers({offer.id()}, {CREATE(persistentVolume)});
  
    AWAIT_READY(createOperationStatus);
    EXPECT_EQ(OPERATION_FAILED, createOperationStatus->status().state());
  
    AWAIT_EXPECT_READY(offers)
      << "Failed to wait for an offer containing resource '" << created << "'";
  }
  
  
  // This test verifies that if a persistent volumes is never published by the
  // storage local resource provider, the volume can be destroyed.
  //
  // To accomplish this:
  //   1. Create a MOUNT disk from a RAW disk resource.
  //   2. Create a persistent volume on the MOUNT disk then launches a task to
  //      write a file into it.
  //   3. Return `UNIMPLEMENTED` for the `NodePublishVolume` call. The task will
  //      fail to launch.
  //   4. Destroy the persistent volume and the MOUNT disk.
  TEST_P(StorageLocalResourceProviderTest, DestroyUnpublishedPersistentVolume)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    const string mockCsiEndpoint =
      "unix://" + path::join(sandbox.get(), "mock_csi.sock");
  
    MockCSIPlugin plugin;
    ASSERT_SOME(plugin.startup(mockCsiEndpoint));
  
    setupResourceProviderConfig(Bytes(0), None(), mockCsiEndpoint);
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Create a persistent MOUNT volume then launch a task to write a file.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<Nothing> taskFailed;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FAILED)))
      .WillOnce(FutureSatisfy(&taskFailed));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    // Fail resource publishing.
    Future<Nothing> nodePublishVolumeCall = futureNodePublishVolumeCall(
        &plugin, StatusError(grpc::Status(grpc::UNIMPLEMENTED, "")));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("touch " + path::join("volume", "file")))})});
  
    AWAIT_READY(nodePublishVolumeCall);
  
    AWAIT_READY(taskFailed);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the persistent volume and the MOUNT disk.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateOperationStatusMessage> destroyDiskOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
    Future<UpdateOperationStatusMessage> destroyOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {DESTROY(persistentVolume), DESTROY_DISK(created)});
  
    AWAIT_READY(destroyOperationStatus);
    EXPECT_EQ(OPERATION_FINISHED, destroyOperationStatus->status().state());
  
    AWAIT_READY(destroyDiskOperationStatus);
    EXPECT_EQ(OPERATION_FINISHED, destroyDiskOperationStatus->status().state());
  
    AWAIT_EXPECT_READY(offers)
      << "Failed to wait for an offer containing resource '" << raw << "'";
  }
  
  
  // This test verifies that if a persistent volumes is never published by the
  // storage local resource provider, the volume can be destroyed after recovery.
  //
  // To accomplish this:
  //   1. Create a MOUNT disk from a RAW disk resource.
  //   2. Create a persistent volume on the MOUNT disk then launches a task to
  //      write a file into it.
  //   3. Return `UNIMPLEMENTED` for the `NodePublishVolume` call. The task will
  //      fail to launch.
  //   4. Restart the agent.
  //   5. Destroy the persistent volume and the MOUNT disk.
  TEST_P(
      StorageLocalResourceProviderTest,
      DestroyUnpublishedPersistentVolumeWithRecovery)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    const string mockCsiEndpoint =
      "unix://" + path::join(sandbox.get(), "mock_csi.sock");
  
    MockCSIPlugin plugin;
    ASSERT_SOME(plugin.startup(mockCsiEndpoint));
  
    setupResourceProviderConfig(Bytes(0), None(), mockCsiEndpoint);
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Create a persistent MOUNT volume then launch a task to write a file.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<Nothing> taskFailed;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FAILED)))
      .WillOnce(FutureSatisfy(&taskFailed));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    // Fail resource publishing.
    Future<Nothing> nodePublishVolumeCall = futureNodePublishVolumeCall(
        &plugin, StatusError(grpc::Status(grpc::UNIMPLEMENTED, "")));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("touch " + path::join("volume", "file")))})});
  
    AWAIT_READY(nodePublishVolumeCall);
  
    AWAIT_READY(taskFailed);
  
    AWAIT_READY(offers);
  
    // Restart the agent.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave.get()->terminate();
  
    // NOTE: The mock CSI plugin always returns 4GB for `GetCapacity` calls, hence
    // the resource provider would report an extra storage pool, and when it shows
    // up, we know that the resource provider has finished reconciling storage
    // pools and thus operations won't be dropped.
    //
    // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
    // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
    // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
    // resource provider reregistration) and wait for the third one, which should
    // contain the extra storage pool. We let it fall through to trigger an offer
    // allocation for the persistent volume.
    //
    // Since the extra storage pool is never used, we reject the offers if only
    // the storage pool is presented.
    //
    // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // NOTE: The order of these expectations is reversed because Google Mock will
    // search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlaveMessage =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
    FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(updateSlaveMessage);
    ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
    ASSERT_FALSE(Resources(
        updateSlaveMessage->resource_providers().providers(0).total_resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .empty());
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the persistent volume and the MOUNT disk.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateOperationStatusMessage> destroyDiskOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
    Future<UpdateOperationStatusMessage> destroyOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {DESTROY(persistentVolume), DESTROY_DISK(created)});
  
    AWAIT_READY(destroyOperationStatus);
    EXPECT_EQ(OPERATION_FINISHED, destroyOperationStatus->status().state());
  
    AWAIT_READY(destroyDiskOperationStatus);
    EXPECT_EQ(OPERATION_FINISHED, destroyDiskOperationStatus->status().state());
  
    AWAIT_EXPECT_READY(offers)
      << "Failed to wait for an offer containing resource '" << raw << "'";
  }
  
  
  // This test verifies that if a persistent volumes is never published by the
  // storage local resource provider, the volume can be destroyed after agent
  // reboot.
  //
  // To accomplish this:
  //   1. Create a MOUNT disk from a RAW disk resource.
  //   2. Create a persistent volume on the MOUNT disk then launches a task to
  //      write a file into it.
  //   3. Return `UNIMPLEMENTED` for the `NodePublishVolume` call. The task will
  //      fail to launch.
  //   4. Simulate an agent reboot.
  //   5. Destroy the persistent volume and the MOUNT disk.
  TEST_P(
      StorageLocalResourceProviderTest,
      DestroyUnpublishedPersistentVolumeWithReboot)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    const string mockCsiEndpoint =
      "unix://" + path::join(sandbox.get(), "mock_csi.sock");
  
    MockCSIPlugin plugin;
    ASSERT_SOME(plugin.startup(mockCsiEndpoint));
  
    setupResourceProviderConfig(Bytes(0), None(), mockCsiEndpoint);
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Create a persistent MOUNT volume then launch a task to write a file.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<Nothing> taskFailed;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FAILED)))
      .WillOnce(FutureSatisfy(&taskFailed));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    // Fail resource publishing.
    Future<Nothing> nodePublishVolumeCall = futureNodePublishVolumeCall(
        &plugin, StatusError(grpc::Status(grpc::UNIMPLEMENTED, "")));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("touch " + path::join("volume", "file")))})});
  
    AWAIT_READY(nodePublishVolumeCall);
  
    AWAIT_READY(taskFailed);
  
    AWAIT_READY(offers);
  
    // Shutdown the agent and unmount all CSI volumes to simulate a reboot.
    EXPECT_CALL(sched, offerRescinded(_, _));
  
    slave->reset();
  
    const string csiRootDir = slave::paths::getCsiRootDir(slaveFlags.work_dir);
    ASSERT_SOME(fs::unmountAll(csiRootDir));
  
    // Inject the boot IDs to simulate a reboot.
    ASSERT_SOME(os::write(
        slave::paths::getBootIdPath(
            slave::paths::getMetaRootDir(slaveFlags.work_dir)),
        "rebooted! ;)"));
  
    Try<list<string>> volumePaths =
      csi::paths::getVolumePaths(csiRootDir, "*", "*");
    ASSERT_SOME(volumePaths);
    ASSERT_FALSE(volumePaths->empty());
  
    foreach (const string& path, volumePaths.get()) {
      Try<csi::paths::VolumePath> volumePath =
        csi::paths::parseVolumePath(csiRootDir, path);
      ASSERT_SOME(volumePath);
  
      const string volumeStatePath = csi::paths::getVolumeStatePath(
          csiRootDir, volumePath->type, volumePath->name, volumePath->volumeId);
  
      Result<csi::state::VolumeState> volumeState =
        slave::state::read<csi::state::VolumeState>(volumeStatePath);
  
      ASSERT_SOME(volumeState);
  
      if (volumeState->state() == csi::state::VolumeState::PUBLISHED) {
        volumeState->set_boot_id("rebooted! ;)");
        ASSERT_SOME(slave::state::checkpoint(volumeStatePath, volumeState.get()));
      }
    }
  
    // NOTE: The mock CSI plugin always returns 4GB for `GetCapacity` calls, hence
    // the resource provider would report an extra storage pool, and when it shows
    // up, we know that the resource provider has finished reconciling storage
    // pools and thus operations won't be dropped.
    //
    // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
    // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
    // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
    // resource provider reregistration) and wait for the third one, which should
    // contain the extra storage pool. We let it fall through to trigger an offer
    // allocation for the persistent volume.
    //
    // Since the extra storage pool is never used, we reject the offers if only
    // the storage pool is presented.
    //
    // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // NOTE: The order of these expectations is reversed because Google Mock will
    // search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlaveMessage =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
    FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
  
    // Restart the agent.
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(updateSlaveMessage);
    ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
    ASSERT_FALSE(Resources(
        updateSlaveMessage->resource_providers().providers(0).total_resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .empty());
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the persistent volume and the MOUNT disk.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateOperationStatusMessage> destroyDiskOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
    Future<UpdateOperationStatusMessage> destroyOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(raw)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {DESTROY(persistentVolume), DESTROY_DISK(created)});
  
    AWAIT_READY(destroyOperationStatus);
    EXPECT_EQ(OPERATION_FINISHED, destroyOperationStatus->status().state());
  
    AWAIT_READY(destroyDiskOperationStatus);
    EXPECT_EQ(OPERATION_FINISHED, destroyDiskOperationStatus->status().state());
  
    AWAIT_EXPECT_READY(offers)
      << "Failed to wait for an offer containing resource '" << raw << "'";
  }
  
  
  // This test verifies that if the storage local resource provider fails to clean
  // up a persistent volume, the volume will not be destroyed.
  //
  // To accomplish this:
  //   1. Creates a MOUNT disk from a RAW disk resource.
  //   2. Creates a persistent volume on the MOUNT disk then launches a task to
  //      write a file into it.
  //   3. Bind-mounts a file in the CSI volume so the persistent volume cleanup
  //      would fail with EBUSY.
  //   4. Destroys the persistent volume and the MOUNT disk. `DESTROY` would fail
  //      and `DESTROY_DISK` would be dropped.
  TEST_P(
      StorageLocalResourceProviderTest, ROOT_DestroyPersistentMountVolumeFailed)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers. The master can send such offers before the
    // resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Create a persistent MOUNT volume then launch a task to write a file.
    Resource persistentVolume = created;
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_id(id::UUID::random().toString());
    persistentVolume.mutable_disk()->mutable_persistence()
      ->set_principal(framework.principal());
    persistentVolume.mutable_disk()->mutable_volume()
      ->set_container_path("volume");
    persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
  
    Future<Nothing> taskFinished;
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
    EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
      .WillOnce(FutureSatisfy(&taskFinished));
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE(persistentVolume),
         LAUNCH({createTask(
             offer.slave_id(),
             createTaskResources(persistentVolume),
             createCommandInfo("touch " + path::join("volume", "file")))})});
  
    AWAIT_READY(taskFinished);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Bind-mount the file in the CSI volume to fail persistent volume cleanup.
    const string& volumePath = created.disk().source().id();
    const string filePath = path::join(volumePath, "file");
    ASSERT_TRUE(os::exists(filePath));
    ASSERT_SOME(fs::mount(filePath, filePath, None(), MS_BIND, None()));
  
    // Destroy the persistent volume and the MOUNT disk, but none will succeed.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateOperationStatusMessage> destroyDiskOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
    Future<UpdateOperationStatusMessage> destroyOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    EXPECT_CALL(
        sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()}, {DESTROY(persistentVolume), DESTROY_DISK(created)});
  
    AWAIT_READY(destroyOperationStatus);
    EXPECT_EQ(OPERATION_FAILED, destroyOperationStatus->status().state());
  
    AWAIT_READY(destroyDiskOperationStatus);
    EXPECT_EQ(OPERATION_DROPPED, destroyDiskOperationStatus->status().state());
  
    AWAIT_READY(offers);
  
    // Check if the MOUNT disk still exists and not being cleaned up.
    EXPECT_TRUE(os::exists(filePath));
  }
  
  
  // This test verifies that the storage local resource provider can import a
  // preprovisioned CSI volume as a MOUNT disk of a given profile if the profile
  // is known to the disk profile adaptor, and can return the space back to the
  // storage pool through either destroying the MOUNT disk, or destroying a RAW
  // disk directly.
  TEST_P(StorageLocalResourceProviderTest, CreateDestroyPreprovisionedVolume)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    // NOTE: We set up the resource provider with an extra storage pool, so that
    // when the storage pool shows up, we know that the resource provider has
    // finished reconciling storage pools and thus operations won't be dropped.
    //
    // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
    // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
    // `UpdateSlaveMessage`s (one sent after agent registration and one after
    // resource provider registration) and wait for the third one, which should
    // contain the extra storage pool. We let it fall through to trigger an offer
    // allocation for the persistent volume.
    //
    // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
    setupResourceProviderConfig(Gigabytes(1), "volume1:2GB;volume2:2GB");
  
    // NOTE: The order of these expectations is reversed because Google Mock will
    // search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlaveMessage =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
    DROP_PROTOBUFS(SlaveRegisteredMessage(), _, _);
    FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(updateSlaveMessage);
    ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
    ASSERT_FALSE(Resources(
        updateSlaveMessage->resource_providers().providers(0).total_resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .empty());
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have wanted
    // resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline unwanted offers, e.g., offers containing only agent-default
    // resources and/or the extra storage pool.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    // We first wait for an offer containing the two preprovisioned volumes to
    // exercise two `CREATE_DISK` operations. Since one of them would fail, we
    // might get subsequent offers containing a preprovisioned volumes after the
    // operations are exercised, so we decline them.
    Future<vector<Offer>> offers;
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        isPreprovisionedVolume<Resource>)))
      .WillOnce(FutureArg<1>(&offers))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    vector<Resource> preprovisioned = google::protobuf::convert<Resource>(
        Resources(offer.resources()).filter(isPreprovisionedVolume<Resource>));
  
    ASSERT_EQ(2u, preprovisioned.size());
  
    // Get the volume paths of the preprovisioned volumes.
    vector<string> volumePaths;
    foreach (const Resource& volume, preprovisioned) {
      const string& volumePath = volume.disk().source().id();
      ASSERT_TRUE(os::exists(volumePath));
      volumePaths.push_back(volumePath);
    }
  
    // Apply profile 'test' and 'unknown' to the preprovisioned volumes. The first
    // operation would succeed but the second would fail.
    Future<multiset<OperationState>> createDiskOperationStates =
      process::collect(
          FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _),
          FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _))
        .then([](const std::tuple<
                  UpdateOperationStatusMessage,
                  UpdateOperationStatusMessage>& operationStatuses) {
          return multiset<OperationState>{
            std::get<0>(operationStatuses).status().state(),
            std::get<1>(operationStatuses).status().state()};
        });
  
    // Once both operations are completed, we would get an offer containing both a
    // MOUNT disk and a preprovisioned volume.
    EXPECT_CALL(sched, resourceOffers(&driver, AllOf(
        OffersHaveAnyResource(
            std::bind(isMountDisk<Resource>, lambda::_1, "test")),
        OffersHaveResource(preprovisioned[1]))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {CREATE_DISK(
             preprovisioned[0], Resource::DiskInfo::Source::MOUNT, "test"),
         CREATE_DISK(
             preprovisioned[1], Resource::DiskInfo::Source::MOUNT, "unknown")});
  
    multiset<OperationState> expectedOperationStates = {
      OperationState::OPERATION_FINISHED,
      OperationState::OPERATION_FAILED};
  
    AWAIT_EXPECT_EQ(expectedOperationStates, createDiskOperationStates);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    Resource imported = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    // Destroy the imported and unimported volumes.
    Future<multiset<OperationState>> destroyDiskOperationStates =
      process::collect(
          FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _),
          FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _))
        .then([](const std::tuple<
                  UpdateOperationStatusMessage,
                  UpdateOperationStatusMessage>& operationStatuses) {
          return multiset<OperationState>{
            std::get<0>(operationStatuses).status().state(),
            std::get<1>(operationStatuses).status().state()};
        });
  
    // Once both operations are completed, we would get an offer containing a
    // storage pool that has all the freed space.
    EXPECT_CALL(sched, resourceOffers(&driver,
        OffersHaveAnyResource([](const Resource& r) {
          return isStoragePool(r, "test") &&
            Megabytes(r.scalar().value()) >= Gigabytes(4);
        })))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.acceptOffers(
        {offer.id()},
        {DESTROY_DISK(imported),
         DESTROY_DISK(preprovisioned[1])});
  
    expectedOperationStates = {
      OperationState::OPERATION_FINISHED,
      OperationState::OPERATION_FINISHED};
  
    AWAIT_EXPECT_EQ(expectedOperationStates, destroyDiskOperationStates);
  
    AWAIT_READY(offers);
  
    // Check if the volume is deleted by the test CSI plugin.
    foreach (const string& volumePath, volumePaths) {
      EXPECT_FALSE(os::exists(volumePath));
    }
  }
  
  
  // This test verifies that operation status updates are resent to the master
  // after being dropped en route to it.
  //
  // To accomplish this:
  //   1. Creates a volume from a RAW disk resource.
  //   2. Drops the first `UpdateOperationStatusMessage` from the agent to the
  //      master, so that it isn't acknowledged by the master.
  //   3. Advances the clock and verifies that the agent resends the operation
  //      status update.
  TEST_P(StorageLocalResourceProviderTest, RetryOperationStatusUpdate)
  {
    Clock::pause();
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags flags = CreateSlaveFlags();
    flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), flags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(flags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Clock::pause();
  
    // Register a framework to exercise an operation.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    Future<vector<Offer>> offers;
  
    // Decline offers without RAW disk resources. The master can send such offers
    // before the resource provider reports its RAW resources, or after receiving
    // the `UpdateOperationStatusMessage`.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers());
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    // NOTE: If the framework has not declined an unwanted offer yet when the
    // resource provider reports its RAW resources, the new allocation triggered
    // by this update won't generate an allocatable offer due to no CPU and memory
    // resources. So we first settle the clock to ensure that the unwanted offer
    // has been declined, then advance the clock to trigger another allocation.
    Clock::settle();
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_FALSE(offers->empty());
  
    const Offer& offer = offers->at(0);
  
    Option<Resource> source;
    foreach (const Resource& resource, offer.resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
        break;
      }
    }
  
    ASSERT_SOME(source);
  
    // We'll drop the first operation status update from the agent to the master.
    Future<UpdateOperationStatusMessage> droppedUpdateOperationStatusMessage =
      DROP_PROTOBUF(
          UpdateOperationStatusMessage(), slave.get()->pid, master.get()->pid);
  
    // Create a volume.
    driver.acceptOffers(
        {offer.id()},
        {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(droppedUpdateOperationStatusMessage);
  
    // The SLRP should resend the dropped operation status update after the
    // status update retry interval minimum.
    Future<UpdateOperationStatusMessage> retriedUpdateOperationStatusMessage =
      FUTURE_PROTOBUF(
          UpdateOperationStatusMessage(), slave.get()->pid, master.get()->pid);
  
    // The master should acknowledge the operation status update.
    Future<AcknowledgeOperationStatusMessage> acknowledgeOperationStatusMessage =
      FUTURE_PROTOBUF(
        AcknowledgeOperationStatusMessage(), master.get()->pid, slave.get()->pid);
  
    // Since the acknowledgement is delivered to the SLRP via HTTP, we wait for
    // a dispatch event to ensure that the acknowledgement is received by SLRP.
    Future<Nothing> statusUpdateManagerAcknowledgement = FUTURE_DISPATCH(
        _,
        (&StatusUpdateManagerProcess<
            id::UUID,
            UpdateOperationStatusRecord,
            UpdateOperationStatusMessage>::acknowledgement));
  
    Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
  
    AWAIT_READY(retriedUpdateOperationStatusMessage);
  
    AWAIT_READY(acknowledgeOperationStatusMessage);
    AWAIT_READY(statusUpdateManagerAcknowledgement);
  
    // The master acknowledged the operation status update, so the SLRP shouldn't
    // send further operation status updates.
    EXPECT_NO_FUTURE_PROTOBUFS(UpdateOperationStatusMessage(), _, _);
  
    Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
    Clock::settle();
  
    driver.stop();
    driver.join();
  }
  
  
  // This test verifies that on agent restarts, unacknowledged operation status
  // updates are resent to the master
  //
  // To accomplish this:
  //   1. Creates a volume from a RAW disk resource.
  //   2. Drops the first `UpdateOperationStatusMessage` from the agent to the
  //      master, so that it isn't acknowledged by the master.
  //   3. Restarts the agent.
  //   4. Verifies that the agent resends the operation status update.
  TEST_P(
      StorageLocalResourceProviderTest,
      RetryOperationStatusUpdateAfterRecovery)
  {
    Clock::pause();
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags flags = CreateSlaveFlags();
    flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), flags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(flags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Clock::pause();
  
    // Register a framework to exercise an operation.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    Future<vector<Offer>> offers;
  
    // Decline offers without RAW disk resources. The master can send such offers
    // before the resource provider reports its RAW resources, or after receiving
    // the `UpdateOperationStatusMessage`.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers());
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    // NOTE: If the framework has not declined an unwanted offer yet when the
    // resource provider reports its RAW resources, the new allocation triggered
    // by this update won't generate an allocatable offer due to no CPU and memory
    // resources. So we first settle the clock to ensure that the unwanted offer
    // has been declined, then advance the clock to trigger another allocation.
    Clock::settle();
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_FALSE(offers->empty());
  
    const Offer& offer = offers->at(0);
  
    Option<Resource> source;
    foreach (const Resource& resource, offer.resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
        break;
      }
    }
  
    ASSERT_SOME(source);
  
    // We'll drop the first operation status update from the agent to the master.
    Future<UpdateOperationStatusMessage> droppedUpdateOperationStatusMessage =
      DROP_PROTOBUF(
          UpdateOperationStatusMessage(), slave.get()->pid, master.get()->pid);
  
    // Create a volume.
    driver.acceptOffers(
        {offer.id()},
        {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(droppedUpdateOperationStatusMessage);
  
    // Restart the agent.
    slave.get()->terminate();
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    Future<UpdateSlaveMessage> updateSlave4 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave3 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    // Once the agent is restarted, the SLRP should resend the dropped operation
    // status update.
    Future<UpdateOperationStatusMessage> retriedUpdateOperationStatusMessage =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, master.get()->pid);
  
    // The master should acknowledge the operation status update once.
    Future<AcknowledgeOperationStatusMessage> acknowledgeOperationStatusMessage =
      FUTURE_PROTOBUF(AcknowledgeOperationStatusMessage(), master.get()->pid, _);
  
    // Since the acknowledgement is delivered to the SLRP via HTTP, we wait for
    // a dispatch event to ensure that the acknowledgement is received by SLRP.
    Future<Nothing> statusUpdateManagerAcknowledgement = FUTURE_DISPATCH(
        _,
        (&StatusUpdateManagerProcess<
            id::UUID,
            UpdateOperationStatusRecord,
            UpdateOperationStatusMessage>::acknowledgement));
  
    slave = StartSlave(detector.get(), flags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(flags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave3);
  
    // Resume the clock so that the CSI plugin's standalone container is created
    // and the SLRP's async loop notices it.
    Clock::resume();
  
    AWAIT_READY(updateSlave4);
    ASSERT_TRUE(updateSlave4->has_resource_providers());
    ASSERT_EQ(1, updateSlave4->resource_providers().providers_size());
  
    Clock::pause();
  
    AWAIT_READY(retriedUpdateOperationStatusMessage);
  
    AWAIT_READY(acknowledgeOperationStatusMessage);
    AWAIT_READY(statusUpdateManagerAcknowledgement);
  
    // The master has acknowledged the operation status update, so the SLRP
    // shouldn't send further operation status updates.
    EXPECT_NO_FUTURE_PROTOBUFS(UpdateOperationStatusMessage(), _, _);
  
    Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
    Clock::settle();
  
    driver.stop();
    driver.join();
  }
  
  
  // This test verifies that storage local resource provider properly
  // reports the metric related to CSI plugin container terminations.
  TEST_P(StorageLocalResourceProviderTest, ContainerTerminationMetric)
  {
    // Since we want to observe a fixed number of `UpdateSlaveMessage`s,
    // register the agent with paused clock to ensure it does not
    // register multiple times due to hitting timeouts.
    Clock::pause();
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
  
    slave::Fetcher fetcher(slaveFlags);
  
    Try<slave::MesosContainerizer*> _containerizer =
      slave::MesosContainerizer::create(slaveFlags, false, &fetcher);
  
    ASSERT_SOME(_containerizer);
  
    Owned<slave::MesosContainerizer> containerizer(_containerizer.get());
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains information from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(
        detector.get(),
        containerizer.get(),
        slaveFlags);
  
    ASSERT_SOME(slave);
  
    Clock::advance(slaveFlags.registration_backoff_factor);
    Clock::advance(slaveFlags.authentication_backoff_factor);
  
    // Wait for the the resource provider to subscribe before killing it
    // by observing the agent report the resource provider to the
    // master. This prevents the plugin from entering a failed, non-
    // restartable state, see MESOS-9130.
    //
    // TODO(bbannier): This step can be removed once MESOS-8400 is implemented.
    AWAIT_READY(updateSlave1);
  
    // Resume the clock so that the CSI plugin's standalone container is created
    // and the SLRP's async loop notices it.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_FALSE(updateSlave2->resource_providers().providers().empty());
  
    Clock::pause();
  
    JSON::Object snapshot = Metrics();
  
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "csi_plugin/container_terminations")));
    EXPECT_EQ(0, snapshot.values.at(metricName(
        "csi_plugin/container_terminations")));
  
    // Get the ID of the CSI plugin container.
    Future<hashset<ContainerID>> pluginContainers = containerizer->containers();
  
    AWAIT_READY(pluginContainers);
    ASSERT_EQ(1u, pluginContainers->size());
  
    const ContainerID& pluginContainerId = *pluginContainers->begin();
  
    Future<Nothing> pluginRestarted =
      FUTURE_DISPATCH(_, &ContainerDaemonProcess::launchContainer);
  
    // Kill the plugin container and wait for it to restart.
    Future<int> pluginKilled = containerizer->status(pluginContainerId)
      .then([](const ContainerStatus& status) {
        return os::kill(status.executor_pid(), SIGKILL);
      });
  
    AWAIT_ASSERT_EQ(0, pluginKilled);
  
    // Resume the clock so the plugin container is properly destroyed.
    Clock::resume();
  
    AWAIT_READY(pluginRestarted);
  
    Clock::pause();
  
    snapshot = Metrics();
  
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "csi_plugin/container_terminations")));
    EXPECT_EQ(1, snapshot.values.at(metricName(
        "csi_plugin/container_terminations")));
  }
  
  
  // This test verifies that operation status updates contain the
  // agent ID and resource provider ID of originating providers.
  TEST_P(StorageLocalResourceProviderTest, OperationUpdate)
  {
    Clock::pause();
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags flags = CreateSlaveFlags();
    flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), flags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(flags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Clock::pause();
  
    // Register a framework to exercise an operation.
    v1::FrameworkInfo frameworkInfo = v1::DEFAULT_FRAMEWORK_INFO;
    frameworkInfo.set_roles(0, "storage");
  
    auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
  
    EXPECT_CALL(*scheduler, connected(_))
      .WillOnce(v1::scheduler::SendSubscribe(frameworkInfo));
  
    Future<v1::scheduler::Event::Subscribed> subscribed;
    EXPECT_CALL(*scheduler, subscribed(_, _))
      .WillOnce(FutureArg<1>(&subscribed));
  
    EXPECT_CALL(*scheduler, heartbeat(_))
      .WillRepeatedly(Return()); // Ignore heartbeats.
  
    // Decline offers that do not contain wanted resources.
    EXPECT_CALL(*scheduler, offers(_, _))
      .WillRepeatedly(v1::scheduler::DeclineOffers());
  
    Future<v1::scheduler::Event::Offers> offers;
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(
            std::bind(isStoragePool<v1::Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    v1::scheduler::TestMesos mesos(
        master.get()->pid,
        ContentType::PROTOBUF,
        scheduler);
  
    AWAIT_READY(subscribed);
  
    const v1::FrameworkID& frameworkId = subscribed->framework_id();
  
    // NOTE: If the framework has not declined an unwanted offer yet when
    // the master updates the agent with the RAW disk resource, the new
    // allocation triggered by this update won't generate an allocatable
    // offer due to no CPU and memory resources. So here we first settle
    // the clock to ensure that the unwanted offer has been declined, then
    // advance the clock to trigger another allocation.
    Clock::settle();
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_FALSE(offers->offers().empty());
  
    const v1::Offer& offer = offers->offers(0);
  
    const v1::AgentID& agentId = offer.agent_id();
  
    Option<v1::Resource> source;
    Option<mesos::v1::ResourceProviderID> resourceProviderId;
    foreach (const v1::Resource& resource, offer.resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
  
        ASSERT_TRUE(resource.has_provider_id());
        resourceProviderId = resource.provider_id();
  
        break;
      }
    }
  
    ASSERT_SOME(source);
    ASSERT_SOME(resourceProviderId);
  
    Future<v1::scheduler::Event::UpdateOperationStatus> update;
  
    EXPECT_CALL(*scheduler, updateOperationStatus(_, _))
      .WillOnce(FutureArg<1>(&update))
      .WillRepeatedly(Return()); // Ignore subsequent updates.
  
    // Create a volume.
    v1::OperationID operationId;
    operationId.set_value("operation");
  
    mesos.send(v1::createCallAccept(
        frameworkId,
        offer,
        {v1::CREATE_DISK(
             source.get(),
             v1::Resource::DiskInfo::Source::MOUNT,
             None(),
             operationId)}));
  
    AWAIT_READY(update);
  
    ASSERT_EQ(operationId, update->status().operation_id());
    ASSERT_EQ(
        mesos::v1::OperationState::OPERATION_FINISHED, update->status().state());
    ASSERT_TRUE(update->status().has_uuid());
    EXPECT_TRUE(metricEquals("master/operations/finished", 1));
  
    ASSERT_TRUE(update->status().has_agent_id());
    EXPECT_EQ(agentId, update->status().agent_id());
  
    ASSERT_TRUE(update->status().has_resource_provider_id());
    EXPECT_EQ(resourceProviderId.get(), update->status().resource_provider_id());
  }
  
  
  // This test verifies that storage local resource provider properly
  // reports metrics related to operation states.
  // TODO(chhsiao): Currently there is no way to test the `pending` metric for
  // operations since we have no control over the completion of an operation. Once
  // we support out-of-band CSI plugins through domain sockets, we could test this
  // metric against a mock CSI plugin.
  TEST_P(StorageLocalResourceProviderTest, OperationStateMetrics)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // The framework is expected to see the following offers in sequence:
    //   1. One containing a RAW disk resource before `CREATE_DISK`.
    //   2. One containing a MOUNT disk resource after `CREATE_DISK`.
    //   3. One containing the same MOUNT disk resource after a failed
    //      `DESTROY_DISK`.
    //
    // We set up the expectations for these offers as the test progresses.
    Future<vector<Offer>> rawDiskOffers;
    Future<vector<Offer>> volumeCreatedOffers;
    Future<vector<Offer>> operationFailedOffers;
  
    Sequence offers;
  
    // We use the following filter to filter offers that do not have
    // wanted resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline offers that contain only the agent's default resources.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&rawDiskOffers));
  
    driver.start();
  
    AWAIT_READY(rawDiskOffers);
    ASSERT_FALSE(rawDiskOffers->empty());
  
    Option<Resource> source;
  
    foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
        break;
      }
    }
  
    ASSERT_SOME(source);
  
    JSON::Object snapshot = Metrics();
  
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/create_disk/finished")));
    EXPECT_EQ(0, snapshot.values.at(metricName(
        "operations/create_disk/finished")));
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/destroy_disk/failed")));
    EXPECT_EQ(0, snapshot.values.at(metricName(
        "operations/destroy_disk/failed")));
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/destroy_disk/dropped")));
    EXPECT_EQ(0, snapshot.values.at(metricName(
        "operations/destroy_disk/dropped")));
  
    // Create a volume.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&volumeCreatedOffers));
  
    driver.acceptOffers(
        {rawDiskOffers->at(0).id()},
        {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(volumeCreatedOffers);
    ASSERT_FALSE(volumeCreatedOffers->empty());
  
    Option<Resource> volume;
  
    foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
      if (isMountDisk(resource, "test")) {
        volume = resource;
        break;
      }
    }
  
    ASSERT_SOME(volume);
    ASSERT_TRUE(volume->disk().source().has_vendor());
    EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
    ASSERT_TRUE(volume->disk().source().has_id());
    ASSERT_TRUE(volume->disk().source().has_mount());
    ASSERT_TRUE(volume->disk().source().mount().has_root());
    EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
  
    snapshot = Metrics();
  
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/create_disk/finished")));
    EXPECT_EQ(1, snapshot.values.at(metricName(
        "operations/create_disk/finished")));
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/destroy_disk/failed")));
    EXPECT_EQ(0, snapshot.values.at(metricName(
        "operations/destroy_disk/failed")));
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/destroy_disk/dropped")));
    EXPECT_EQ(0, snapshot.values.at(metricName(
        "operations/destroy_disk/dropped")));
  
    // Remove the volume out of band to fail `DESTROY_DISK`.
    const string& volumePath = volume->disk().source().id();
    ASSERT_SOME(os::rmdir(volumePath));
  
    // Destroy the created volume, which will fail.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(volume.get())))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&operationFailedOffers))
      .WillRepeatedly(DeclineOffers(declineFilters)); // Decline further offers.
  
    driver.acceptOffers(
        {volumeCreatedOffers->at(0).id()}, {DESTROY_DISK(volume.get())});
  
    AWAIT_READY(operationFailedOffers);
    ASSERT_FALSE(operationFailedOffers->empty());
  
    snapshot = Metrics();
  
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/create_disk/finished")));
    EXPECT_EQ(1, snapshot.values.at(metricName(
        "operations/create_disk/finished")));
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/destroy_disk/failed")));
    EXPECT_EQ(1, snapshot.values.at(metricName(
        "operations/destroy_disk/failed")));
    ASSERT_NE(0u, snapshot.values.count(metricName(
        "operations/destroy_disk/dropped")));
    EXPECT_EQ(0, snapshot.values.at(metricName(
        "operations/destroy_disk/dropped")));
  
    // Destroy the volume again, which will be dropped this time.
    Future<ApplyOperationMessage> applyOperationMessage =
      DROP_PROTOBUF(ApplyOperationMessage(), _, _);
  
    driver.acceptOffers(
        {operationFailedOffers->at(0).id()}, {DESTROY_DISK(volume.get())});
  
    AWAIT_READY(applyOperationMessage);
    ASSERT_TRUE(applyOperationMessage
      ->resource_version_uuid().has_resource_provider_id());
  
    // Modify the resource version UUID to drop `DESTROY_DISK`.
    Future<UpdateOperationStatusMessage> operationDroppedStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    ApplyOperationMessage spoofedApplyOperationMessage =
      applyOperationMessage.get();
    spoofedApplyOperationMessage.mutable_resource_version_uuid()->mutable_uuid()
      ->set_value(id::UUID::random().toBytes());
  
    post(master.get()->pid, slave.get()->pid, spoofedApplyOperationMessage);
  
    AWAIT_READY(operationDroppedStatus);
    EXPECT_EQ(OPERATION_DROPPED, operationDroppedStatus->status().state());
  
    EXPECT_TRUE(metricEquals("master/operations/dropped", 1));
    EXPECT_TRUE(metricEquals("master/operations/create_disk/finished", 1));
    EXPECT_TRUE(metricEquals("master/operations/destroy_disk/failed", 1));
    EXPECT_TRUE(metricEquals("master/operations/destroy_disk/dropped", 1));
  
    EXPECT_TRUE(metricEquals(metricName(
        "operations/create_disk/finished"), 1));
    EXPECT_TRUE(metricEquals(metricName(
        "operations/destroy_disk/failed"), 1));
    EXPECT_TRUE(metricEquals(metricName(
        "operations/destroy_disk/dropped"), 1));
  }
  
  
  // This test verifies that storage local resource provider properly
  // reports metrics related to RPCs to CSI plugins.
  // TODO(chhsiao): Currently there is no way to test the `pending` and
  // `cancelled` metrics for RPCs since we have no control over the completion of
  // an operation. Once we support out-of-band CSI plugins through domain sockets,
  // we could test these metrics against a mock CSI plugin.
  TEST_P(StorageLocalResourceProviderTest, CsiPluginRpcMetrics)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise operations.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // The framework is expected to see the following offers in sequence:
    //   1. One containing a RAW disk resource before `CREATE_DISK`.
    //   2. One containing a MOUNT disk resource after `CREATE_DISK`.
    //   3. One containing the same MOUNT disk resource after a failed
    //      `DESTROY_DISK`.
    //
    // We set up the expectations for these offers as the test progresses.
    Future<vector<Offer>> rawDiskOffers;
    Future<vector<Offer>> volumeCreatedOffers;
    Future<vector<Offer>> operationFailedOffers;
  
    Sequence offers;
  
    // We use the following filter to filter offers that do not have
    // wanted resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline offers that contain only the agent's default resources.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&rawDiskOffers));
  
    driver.start();
  
    AWAIT_READY(rawDiskOffers);
    ASSERT_FALSE(rawDiskOffers->empty());
  
    Option<Resource> source;
  
    foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
        break;
      }
    }
  
    ASSERT_SOME(source);
  
    // We expect that the following RPC calls are made during startup: `Probe`,
    // `GetPluginInfo` (2), `GetCapacity`, `GetPluginCapabilities,
    // `ControllerGetCapabilities`, `ListVolumes` (2), `NodeGetCapabilities`,
    // `NodeGetId`.
    //
    // TODO(chhsiao): As these are implementation details, we should count the
    // calls processed by a mock CSI plugin and check the metrics against that.
    const int numFinishedStartupRpcs = 10;
  
    EXPECT_TRUE(metricEquals(
        metricName("csi_plugin/rpcs_finished"), numFinishedStartupRpcs));
    EXPECT_TRUE(metricEquals(metricName("csi_plugin/rpcs_failed"), 0));
  
    // Create a volume.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&volumeCreatedOffers));
  
    driver.acceptOffers(
        {rawDiskOffers->at(0).id()},
        {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(volumeCreatedOffers);
    ASSERT_FALSE(volumeCreatedOffers->empty());
  
    Option<Resource> volume;
  
    foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
      if (isMountDisk(resource, "test")) {
        volume = resource;
        break;
      }
    }
  
    ASSERT_SOME(volume);
    ASSERT_TRUE(volume->disk().source().has_vendor());
    EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
    ASSERT_TRUE(volume->disk().source().has_id());
    ASSERT_TRUE(volume->disk().source().has_mount());
    ASSERT_TRUE(volume->disk().source().mount().has_root());
    EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
  
    // An additional `CreateVolume` RPC call is now finished.
    EXPECT_TRUE(metricEquals(
        metricName("csi_plugin/rpcs_finished"), numFinishedStartupRpcs + 1));
    EXPECT_TRUE(metricEquals(metricName("csi_plugin/rpcs_failed"), 0));
  
    // Remove the volume out of band to fail `DESTROY_DISK`.
    const string& volumePath = volume->disk().source().id();
    ASSERT_SOME(os::rmdir(volumePath));
  
    // Destroy the created volume, which will fail.
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(volume.get())))
      .InSequence(offers)
      .WillOnce(FutureArg<1>(&operationFailedOffers))
      .WillRepeatedly(DeclineOffers(declineFilters)); // Decline further offers.
  
    driver.acceptOffers(
        {volumeCreatedOffers->at(0).id()}, {DESTROY_DISK(volume.get())});
  
    AWAIT_READY(operationFailedOffers);
    ASSERT_FALSE(operationFailedOffers->empty());
  
    // An additional `DeleteVolume` RPC call is now failed.
    EXPECT_TRUE(metricEquals(
        metricName("csi_plugin/rpcs_finished"), numFinishedStartupRpcs + 1));
    EXPECT_TRUE(metricEquals(metricName("csi_plugin/rpcs_failed"), 1));
  }
  
  
  // Master reconciles operations that are missing from a reregistering slave.
  // In this case, one of the two `ApplyOperationMessage`s is dropped, so the
  // resource provider should send only one OPERATION_DROPPED.
  //
  // TODO(greggomann): Test operations on agent default resources: for such
  // operations, the agent generates the dropped status.
  TEST_P(StorageLocalResourceProviderTest, ReconcileDroppedOperation)
  {
    Clock::pause();
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    StandaloneMasterDetector detector(master.get()->pid);
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    // Since the local resource provider gets subscribed after the agent is
    // registered, it is guaranteed that the agent will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from the
    // storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(&detector, slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by the
    // plugin container, which runs in another Linux process. Since we do not have
    // a `Future` linked to the standalone container launch to await on, it is
    // difficult to accomplish this without resuming the clock.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Clock::pause();
  
    FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
    frameworkInfo.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // We use the following filter to filter offers that do not have
    // wanted resources for 365 days (the maximum).
    Filters declineFilters;
    declineFilters.set_refuse_seconds(Days(365).secs());
  
    // Decline offers that do not contain wanted resources.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers(declineFilters));
  
    Future<vector<Offer>> offersBeforeOperations;
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offersBeforeOperations))
      .WillRepeatedly(DeclineOffers(declineFilters)); // Decline further offers.
  
    driver.start();
  
    AWAIT_READY(offersBeforeOperations);
    ASSERT_EQ(1u, offersBeforeOperations->size());
  
    Resources raw =
      Resources(offersBeforeOperations->at(0).resources())
        .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"));
  
    // Create two MOUNT disks of 2GB each.
    ASSERT_SOME_EQ(Gigabytes(4), raw.disk());
    Resource source1 = *raw.begin();
    source1.mutable_scalar()->set_value(
        static_cast<double>(Gigabytes(2).bytes()) / Bytes::MEGABYTES);
    Resource source2 = *(raw - source1).begin();
  
    // Drop one of the operations on the way to the agent.
    Future<ApplyOperationMessage> applyOperationMessage =
      DROP_PROTOBUF(ApplyOperationMessage(), _, _);
  
    // The successful operation will result in a terminal update.
    Future<UpdateOperationStatusMessage> operationFinishedStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    // Attempt the creation of two volumes.
    driver.acceptOffers(
        {offersBeforeOperations->at(0).id()},
        {CREATE_DISK(source1, Resource::DiskInfo::Source::MOUNT),
         CREATE_DISK(source2, Resource::DiskInfo::Source::MOUNT)});
  
    // Ensure that the operations are processed.
    Clock::settle();
  
    AWAIT_READY(applyOperationMessage);
    AWAIT_READY(operationFinishedStatus);
  
    Future<SlaveReregisteredMessage> slaveReregisteredMessage =
      FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
  
    // Observe explicit operation reconciliation between master and agent.
    Future<ReconcileOperationsMessage> reconcileOperationsMessage =
      FUTURE_PROTOBUF(ReconcileOperationsMessage(), _, _);
    Future<UpdateOperationStatusMessage> operationDroppedStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    // The master may send an offer with the agent's resources after the agent
    // reregisters, but before an `UpdateSlaveMessage` is sent containing the
    // resource provider's resources. In this case, the offer will be rescinded.
    EXPECT_CALL(sched, offerRescinded(&driver, _))
      .Times(AtMost(1));
  
    // Simulate a spurious master change event (e.g., due to ZooKeeper
    // expiration) at the slave to force re-registration.
    detector.appoint(master.get()->pid);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(slaveReregisteredMessage);
    AWAIT_READY(reconcileOperationsMessage);
    AWAIT_READY(operationDroppedStatus);
  
    std::set<OperationState> expectedStates =
      {OperationState::OPERATION_DROPPED,
       OperationState::OPERATION_FINISHED};
  
    std::set<OperationState> observedStates =
      {operationFinishedStatus->status().state(),
       operationDroppedStatus->status().state()};
  
    ASSERT_EQ(expectedStates, observedStates);
  
    Future<vector<Offer>> offersAfterOperations;
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offersAfterOperations));
  
    // Advance the clock to trigger a batch allocation.
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offersAfterOperations);
    ASSERT_FALSE(offersAfterOperations->empty());
  
    Resources converted =
      Resources(offersAfterOperations->at(0).resources())
        .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"));
  
    ASSERT_EQ(1u, converted.size());
  
    // TODO(greggomann): Add inspection of dropped operation metrics here once
    // such metrics have been added. See MESOS-8406.
  
    // Settle the clock to ensure that unexpected messages will cause errors.
    Clock::settle();
  
    driver.stop();
    driver.join();
  }
  
  
  // This test verifies that if an operation ID is specified, operation status
  // updates are resent to the scheduler until acknowledged.
  TEST_P(StorageLocalResourceProviderTest, RetryOperationStatusUpdateToScheduler)
  {
    Clock::pause();
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags flags = CreateSlaveFlags();
    flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider daemon is started after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), flags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(flags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Clock::pause();
  
    // Register a framework to exercise an operation.
    v1::FrameworkInfo frameworkInfo = v1::DEFAULT_FRAMEWORK_INFO;
    frameworkInfo.set_roles(0, "storage");
  
    auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
  
    EXPECT_CALL(*scheduler, connected(_))
      .WillOnce(v1::scheduler::SendSubscribe(frameworkInfo));
  
    Future<v1::scheduler::Event::Subscribed> subscribed;
    EXPECT_CALL(*scheduler, subscribed(_, _))
      .WillOnce(FutureArg<1>(&subscribed));
  
    EXPECT_CALL(*scheduler, heartbeat(_))
      .WillRepeatedly(Return()); // Ignore heartbeats.
  
    // Decline offers that do not contain wanted resources.
    EXPECT_CALL(*scheduler, offers(_, _))
      .WillRepeatedly(v1::scheduler::DeclineOffers());
  
    Future<v1::scheduler::Event::Offers> offers;
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(
            std::bind(isStoragePool<v1::Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    v1::scheduler::TestMesos mesos(
        master.get()->pid,
        ContentType::PROTOBUF,
        scheduler);
  
    AWAIT_READY(subscribed);
    v1::FrameworkID frameworkId(subscribed->framework_id());
  
    // NOTE: If the framework has not declined an unwanted offer yet when
    // the master updates the agent with the RAW disk resource, the new
    // allocation triggered by this update won't generate an allocatable
    // offer due to no CPU and memory resources. So here we first settle
    // the clock to ensure that the unwanted offer has been declined, then
    // advance the clock to trigger another allocation.
    Clock::settle();
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_FALSE(offers->offers().empty());
  
    const v1::Offer& offer = offers->offers(0);
  
    Option<v1::Resource> source;
    Option<mesos::v1::ResourceProviderID> resourceProviderId;
    foreach (const v1::Resource& resource, offer.resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
  
        ASSERT_TRUE(resource.has_provider_id());
        resourceProviderId = resource.provider_id();
  
        break;
      }
    }
  
    ASSERT_SOME(source);
    ASSERT_SOME(resourceProviderId);
  
    Future<v1::scheduler::Event::UpdateOperationStatus> update;
    Future<v1::scheduler::Event::UpdateOperationStatus> retriedUpdate;
  
    EXPECT_CALL(*scheduler, updateOperationStatus(_, _))
      .WillOnce(FutureArg<1>(&update))
      .WillOnce(FutureArg<1>(&retriedUpdate));
  
    // Create a volume.
    v1::OperationID operationId;
    operationId.set_value("operation");
  
    mesos.send(v1::createCallAccept(
        frameworkId,
        offer,
        {v1::CREATE_DISK(
             source.get(),
             v1::Resource::DiskInfo::Source::MOUNT,
             None(),
             operationId)}));
  
    AWAIT_READY(update);
  
    ASSERT_EQ(operationId, update->status().operation_id());
    ASSERT_EQ(
        mesos::v1::OperationState::OPERATION_FINISHED, update->status().state());
    ASSERT_TRUE(update->status().has_uuid());
  
    ASSERT_TRUE(retriedUpdate.isPending());
  
    Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
    Clock::settle();
  
    // The scheduler didn't acknowledge the operation status update, so the SLRP
    // should resend it after the status update retry interval minimum.
    AWAIT_READY(retriedUpdate);
  
    ASSERT_EQ(operationId, retriedUpdate->status().operation_id());
    ASSERT_EQ(
        mesos::v1::OperationState::OPERATION_FINISHED,
        retriedUpdate->status().state());
    ASSERT_TRUE(retriedUpdate->status().has_uuid());
  
    // The scheduler will acknowledge the operation status update, so the agent
    // should receive an acknowledgement.
    Future<AcknowledgeOperationStatusMessage> acknowledgeOperationStatusMessage =
      FUTURE_PROTOBUF(
        AcknowledgeOperationStatusMessage(), master.get()->pid, slave.get()->pid);
  
    // Since the acknowledgement is delivered to the SLRP via HTTP, we wait for
    // a dispatch event to ensure that the acknowledgement is received by SLRP.
    Future<Nothing> statusUpdateManagerAcknowledgement = FUTURE_DISPATCH(
        _,
        (&StatusUpdateManagerProcess<
            id::UUID,
            UpdateOperationStatusRecord,
            UpdateOperationStatusMessage>::acknowledgement));
  
    mesos.send(v1::createCallAcknowledgeOperationStatus(
        frameworkId, offer.agent_id(), resourceProviderId.get(), update.get()));
  
    AWAIT_READY(acknowledgeOperationStatusMessage);
    AWAIT_READY(statusUpdateManagerAcknowledgement);
  
    // Verify that the retry was only counted as one operation.
    EXPECT_TRUE(metricEquals("master/operations/finished", 1));
  
    // Now that the SLRP has received the acknowledgement, the SLRP shouldn't
    // send further operation status updates.
    EXPECT_NO_FUTURE_PROTOBUFS(UpdateOperationStatusMessage(), _, _);
  
    Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
    Clock::settle();
  }
  
  
  // This test ensures that the master responds with the latest state
  // for operations that are terminal at the master, but have not been
  // acknowledged by the framework.
  TEST_P(
      StorageLocalResourceProviderTest,
      ReconcileUnacknowledgedTerminalOperation)
  {
    Clock::pause();
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags flags = CreateSlaveFlags();
    flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), flags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(flags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Clock::pause();
  
    // Register a framework to exercise an operation.
    v1::FrameworkInfo frameworkInfo = v1::DEFAULT_FRAMEWORK_INFO;
    frameworkInfo.set_roles(0, "storage");
  
    auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
  
    EXPECT_CALL(*scheduler, connected(_))
      .WillOnce(v1::scheduler::SendSubscribe(frameworkInfo));
  
    Future<v1::scheduler::Event::Subscribed> subscribed;
    EXPECT_CALL(*scheduler, subscribed(_, _))
      .WillOnce(FutureArg<1>(&subscribed));
  
    EXPECT_CALL(*scheduler, heartbeat(_))
      .WillRepeatedly(Return()); // Ignore heartbeats.
  
    // Decline offers that do not contain wanted resources.
    EXPECT_CALL(*scheduler, offers(_, _))
      .WillRepeatedly(v1::scheduler::DeclineOffers());
  
    Future<v1::scheduler::Event::Offers> offers;
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(
            std::bind(isStoragePool<v1::Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    v1::scheduler::TestMesos mesos(
        master.get()->pid,
        ContentType::PROTOBUF,
        scheduler);
  
    AWAIT_READY(subscribed);
    v1::FrameworkID frameworkId(subscribed->framework_id());
  
    // NOTE: If the framework has not declined an unwanted offer yet when
    // the master updates the agent with the RAW disk resource, the new
    // allocation triggered by this update won't generate an allocatable
    // offer due to no CPU and memory resources. So here we first settle
    // the clock to ensure that the unwanted offer has been declined, then
    // advance the clock to trigger another allocation.
    Clock::settle();
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_FALSE(offers->offers().empty());
  
    const v1::Offer& offer = offers->offers(0);
    const v1::AgentID& agentId = offer.agent_id();
  
    Future<v1::scheduler::Event::UpdateOperationStatus> update;
    EXPECT_CALL(*scheduler, updateOperationStatus(_, _))
      .WillOnce(FutureArg<1>(&update));
  
    Option<v1::Resource> source;
    Option<mesos::v1::ResourceProviderID> resourceProviderId;
    foreach (const v1::Resource& resource, offer.resources()) {
      if (isStoragePool(resource, "test")) {
        source = resource;
  
        ASSERT_TRUE(resource.has_provider_id());
        resourceProviderId = resource.provider_id();
  
        break;
      }
    }
  
    ASSERT_SOME(source);
    ASSERT_SOME(resourceProviderId);
  
    v1::OperationID operationId;
    operationId.set_value("operation");
  
    mesos.send(v1::createCallAccept(
        frameworkId,
        offer,
        {v1::CREATE_DISK(
             source.get(),
             v1::Resource::DiskInfo::Source::MOUNT,
             None(),
             operationId)}));
  
    AWAIT_READY(update);
  
    ASSERT_EQ(operationId, update->status().operation_id());
    ASSERT_EQ(v1::OperationState::OPERATION_FINISHED, update->status().state());
    ASSERT_TRUE(update->status().has_uuid());
    EXPECT_TRUE(metricEquals("master/operations/finished", 1));
  
    v1::scheduler::Call::ReconcileOperations::Operation operation;
    operation.mutable_operation_id()->CopyFrom(operationId);
    operation.mutable_agent_id()->CopyFrom(agentId);
  
    Future<v1::scheduler::Event::UpdateOperationStatus> reconciliationUpdate;
    EXPECT_CALL(*scheduler, updateOperationStatus(_, _))
      .WillOnce(FutureArg<1>(&reconciliationUpdate));
  
    const Future<v1::scheduler::APIResult> result =
      mesos.call({v1::createCallReconcileOperations(frameworkId, {operation})});
  
    AWAIT_READY(result);
  
    // The master should respond with '202 Accepted' and an
    // empty body (response field unset).
    ASSERT_EQ(process::http::Status::ACCEPTED, result->status_code());
    ASSERT_FALSE(result->has_response());
  
    AWAIT_READY(reconciliationUpdate);
  
    EXPECT_EQ(operationId, reconciliationUpdate->status().operation_id());
    EXPECT_EQ(v1::OPERATION_FINISHED, reconciliationUpdate->status().state());
    EXPECT_FALSE(reconciliationUpdate->status().has_uuid());
  }
  
  
  // This test verifies that the storage local resource provider will retry
  // `CreateVolume` and `DeleteVolume` CSI calls with a random exponential backoff
  // upon receiving `DEADLINE_EXCEEDED` or `UNAVAILABLE`.
  //
  // To accomplish this:
  //   1. Creates a MOUNT disk from a RAW disk resource.
  //   2. Returns `DEADLINE_EXCEEDED` or `UNAVAILABLE` for the first n
  //      `CreateVolume` calls, where `n = numRetryableErrors` defined below. The
  //      clock is advanced exponentially to trigger retries.
  //   3. Returns `OK` for the next `CreateVolume` call.
  //   4. Destroys the MOUNT disk.
  //   5. Returns `DEADLINE_EXCEEDED` or `UNAVAILABLE` for the first n
  //      `DeleteVolume` calls, where `n = numRetryableErrors` defined below. The
  //      clock is advanced exponentially to trigger retries.
  //   6. Returns `UNIMPLEMENTED` for the next `DeleteVolume` call to verify that
  //      there is no retry on a non-retryable error.
  TEST_P(StorageLocalResourceProviderTest, RetryRpcWithExponentialBackoff)
  {
    Clock::pause();
  
    // The number of retryable errors to return for each RPC, should be >= 1.
    const size_t numRetryableErrors = 10;
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    const string mockCsiEndpoint =
      "unix://" + path::join(sandbox.get(), "mock_csi.sock");
  
    MockCSIPlugin plugin;
    ASSERT_SOME(plugin.startup(mockCsiEndpoint));
  
    setupResourceProviderConfig(Bytes(0), None(), mockCsiEndpoint);
  
    master::Flags masterFlags = CreateMasterFlags();
  
    // Set the ping timeout to be sufficiently large to avoid agent disconnection.
    masterFlags.max_agent_ping_timeouts = 1000;
  
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags flags = CreateSlaveFlags();
    flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), flags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration.
    Clock::advance(flags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
    ASSERT_EQ(1, updateSlave2->resource_providers().providers_size());
  
    Clock::pause();
  
    // Register a framework to receive offers.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, "storage");
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // Decline offers without RAW disk resources. The master can send such offers
    // before the resource provider receives profile updates.
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers());
  
    Future<vector<Offer>> offers;
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    driver.start();
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    Offer offer = offers->at(0);
  
    // We expect that the following RPC calls are made during startup: `Probe`,
    // `GetPluginInfo` (2), `GetPluginCapabilities, `ControllerGetCapabilities`,
    // `ListVolumes` (2), `GetCapacity`, `NodeGetCapabilities`, `NodeGetId`.
    //
    // TODO(chhsiao): As these are implementation details, we should count the
    // calls processed by a mock CSI plugin and check the metrics against that.
    const int numFinishedStartupRpcs = 10;
  
    EXPECT_TRUE(metricEquals(
        metricName("csi_plugin/rpcs_finished"), numFinishedStartupRpcs));
    EXPECT_TRUE(metricEquals(metricName("csi_plugin/rpcs_failed"), 0));
  
    // Create a MOUNT disk.
    Resource raw = *Resources(offer.resources())
      .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
      .begin();
  
    EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
        std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
      .WillOnce(FutureArg<1>(&offers));
  
    Future<UpdateOperationStatusMessage> updateOperationStatus =
      FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    // Return `DEADLINE_EXCEEDED` for the first `CreateVolume` call.
    Future<Nothing> createVolumeCall = futureCreateVolumeCall(
        &plugin, StatusError(grpc::Status(grpc::DEADLINE_EXCEEDED, "")));
  
    driver.acceptOffers(
        {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
    AWAIT_READY(createVolumeCall) << "Failed to wait for CreateVolume call #1";
  
    Duration createVolumeBackoff = csi::DEFAULT_RPC_RETRY_BACKOFF_FACTOR;
  
    for (size_t i = 1; i < numRetryableErrors; i++) {
      // Return `UNAVAILABLE` for subsequent `CreateVolume` calls.
      createVolumeCall = futureCreateVolumeCall(
          &plugin, StatusError(grpc::Status(grpc::UNAVAILABLE, "")));
  
      // Settle the clock to ensure that the retry timer has been set, then
      // advance the clock by the maximum backoff to trigger a retry.
      Clock::settle();
      Clock::advance(createVolumeBackoff);
  
      AWAIT_READY(createVolumeCall)
        << "Failed to wait for CreateVolume call #" << (i + 1);
  
      createVolumeBackoff = std::min(
          createVolumeBackoff * 2, csi::DEFAULT_RPC_RETRY_INTERVAL_MAX);
    }
  
    // Return a successful response for the last `CreateVolume` call.
    createVolumeCall = futureCreateVolumeCall(
        &plugin,
        csi::VolumeInfo{Megabytes(raw.scalar().value()).bytes(),
                        id::UUID::random().toString()});
  
    // Settle the clock to ensure that the retry timer has been set, then
    // advance the clock by the maximum backoff to trigger a retry.
    Clock::settle();
    Clock::advance(createVolumeBackoff);
  
    AWAIT_READY(createVolumeCall)
      << "Failed to wait for CreateVolume call #" << (numRetryableErrors + 1);
  
    AWAIT_READY(updateOperationStatus);
    EXPECT_EQ(OPERATION_FINISHED, updateOperationStatus->status().state());
  
    // Settle the clock to recover the created disk, then advance the clock to
    // trigger a batch allocation.
    Clock::settle();
    Clock::advance(masterFlags.allocation_interval);
  
    AWAIT_READY(offers);
    ASSERT_EQ(1u, offers->size());
  
    offer = offers->at(0);
  
    // Destroy the MOUNT disk.
    Resource created = *Resources(offer.resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
  
    updateOperationStatus = FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
  
    // Return `DEADLINE_EXCEEDED` for the first `DeleteVolume` call.
    Future<Nothing> deleteVolumeCall = futureDeleteVolumeCall(
        &plugin, StatusError(grpc::Status(grpc::DEADLINE_EXCEEDED, "")));
  
    driver.acceptOffers({offers->at(0).id()}, {DESTROY_DISK(created)});
  
    AWAIT_READY(deleteVolumeCall) << "Failed to wait for DeleteVolume call #1";
  
    Duration deleteVolumeBackoff = csi::DEFAULT_RPC_RETRY_BACKOFF_FACTOR;
  
    for (size_t i = 1; i < numRetryableErrors; i++) {
      // Return `UNAVAILABLE` for subsequent `DeleteVolume` calls.
      deleteVolumeCall = futureDeleteVolumeCall(
          &plugin, StatusError(grpc::Status(grpc::UNAVAILABLE, "")));
  
      // Settle the clock to ensure that the retry timer has been set, then
      // advance the clock by the maximum backoff to trigger a retry.
      Clock::settle();
      Clock::advance(deleteVolumeBackoff);
  
      AWAIT_READY(deleteVolumeCall)
        << "Failed to wait for DeleteVolume call #" << (i + 1);
  
      deleteVolumeBackoff = std::min(
          deleteVolumeBackoff * 2, csi::DEFAULT_RPC_RETRY_INTERVAL_MAX);
    }
  
    // Return a non-retryable error for the last `DeleteVolume` call.
    deleteVolumeCall = futureDeleteVolumeCall(
        &plugin, StatusError(grpc::Status(grpc::UNIMPLEMENTED, "")));
  
    // Settle the clock to ensure that the retry timer has been set, then
    // advance the clock by the maximum backoff to trigger a retry.
    Clock::settle();
    Clock::advance(deleteVolumeBackoff);
  
    AWAIT_READY(deleteVolumeCall)
      << "Failed to wait for DeleteVolume call #" << (numRetryableErrors + 1);
  
    AWAIT_READY(updateOperationStatus);
    EXPECT_EQ(OPERATION_FAILED, updateOperationStatus->status().state());
  
    // Verify that the RPC metrics count the successes and errors correctly.
    //
    // TODO(chhsiao): verify the retry metrics instead once they are in place.
    EXPECT_TRUE(metricEquals("master/operations/failed", 1));
    EXPECT_TRUE(metricEquals("master/operations/finished", 1));
  
    // There should be 1 finished and 10 failed `CreateVolume` calls, and 11
    // failed `DeleteVolume` calls.
    EXPECT_TRUE(metricEquals(
        metricName("csi_plugin/rpcs_finished"), numFinishedStartupRpcs + 1));
    EXPECT_TRUE(metricEquals(
        metricName("csi_plugin/rpcs_failed"), numRetryableErrors * 2 + 1));
  }
  
  
  // This test verifies the master can handle "non-speculative"
  // (i.e. the master cannot assume the success of the operation) operation
  // status updates that originate from frameworks which have been torn down.
  // Non-speculative operations include CREATE_DISK.
  TEST_P(
      StorageLocalResourceProviderTest,
      FrameworkTeardownBeforeTerminalOperationStatusUpdate)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise an operation.
    v1::FrameworkInfo frameworkInfo = v1::DEFAULT_FRAMEWORK_INFO;
    frameworkInfo.set_roles(0, "storage");
  
    auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
  
    // NOTE: The scheduler may connect again after the TEARDOWN call.
    EXPECT_CALL(*scheduler, connected(_))
      .WillOnce(v1::scheduler::SendSubscribe(frameworkInfo))
      .WillRepeatedly(Return());
  
    Future<v1::scheduler::Event::Subscribed> subscribed;
    EXPECT_CALL(*scheduler, subscribed(_, _))
      .WillOnce(FutureArg<1>(&subscribed));
  
    EXPECT_CALL(*scheduler, heartbeat(_))
      .WillRepeatedly(Return()); // Ignore heartbeats.
  
    // Decline offers that contain only the agent's default resources.
    EXPECT_CALL(*scheduler, offers(_, _))
      .WillRepeatedly(v1::scheduler::DeclineOffers());
  
    // We are only interested in an offer with raw disk resources
    auto isRaw = [](const v1::Resource& r) {
      return r.has_disk() &&
        r.disk().has_source() &&
        r.disk().source().has_profile() &&
        r.disk().source().type() == v1::Resource::DiskInfo::Source::RAW;
    };
  
    Future<v1::scheduler::Event::Offers> offers;
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(isRaw)))
      .WillOnce(FutureArg<1>(&offers));
  
    v1::scheduler::TestMesos mesos(
        master.get()->pid,
        ContentType::PROTOBUF,
        scheduler);
  
    AWAIT_READY(subscribed);
  
    const v1::FrameworkID& frameworkId = subscribed->framework_id();
  
    // Prepare to intercept the terminal operation status message between
    // agent and master, to allow the framework to teardown first.
    // This will cause the operation to become an orphan operation.
    Future<UpdateOperationStatusMessage> updateMessage =
      DROP_PROTOBUF(
          UpdateOperationStatusMessage(), slave.get()->pid, master.get()->pid);
  
    AWAIT_READY(offers);
    ASSERT_FALSE(offers->offers().empty());
  
    const v1::Offer& offer = offers->offers(0);
  
    ASSERT_FALSE(offer.resources().empty());
  
    // The test relies on retry logic within the SLRP module to resend the
    // terminal operation feedback message. We pause the clock here to control
    // when this retry happens.
    Clock::pause();
  
    // Have the framework call CREATE_DISK, a non-speculative operation.
    Option<v1::Resource> rawDisk;
  
    foreach (const v1::Resource& resource, offer.resources()) {
      if (isRaw(resource)) {
        rawDisk = resource;
        break;
      }
    }
  
    ASSERT_SOME(rawDisk);
  
    v1::OperationID operationId;
    operationId.set_value("operation");
  
    mesos.send(v1::createCallAccept(
        frameworkId,
        offer,
        {v1::CREATE_DISK(
            rawDisk.get(),
            v1::Resource::DiskInfo::Source::MOUNT,
            None(),
            operationId)}));
  
    // This message will be sent once the agent has completed the operation.
    AWAIT_READY(updateMessage);
  
    EXPECT_EQ(
        OperationState::OPERATION_FINISHED, updateMessage->status().state());
    EXPECT_EQ(
        operationId.value(), updateMessage->status().operation_id().value());
  
    // We can now tear down the framework, since we are certain the operation
    // has reached the agent.
    scheduler.reset();
  
    {
      v1::master::Call v1Call;
      v1Call.set_type(v1::master::Call::TEARDOWN);
  
      v1::master::Call::Teardown* teardown = v1Call.mutable_teardown();
  
      teardown->mutable_framework_id()->CopyFrom(frameworkId);
  
      Future<http::Response> response = process::http::post(
          master.get()->pid,
          "api/v1",
          createBasicAuthHeaders(DEFAULT_CREDENTIAL),
          serialize(ContentType::PROTOBUF, v1Call),
          stringify(ContentType::PROTOBUF));
  
      AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
    }
  
    // Since the scheduler is gone, the master will acknowledge the operation
    // status update on its behalf.
    Future<AcknowledgeOperationStatusMessage> acknowledgeOperationStatusMessage =
      FUTURE_PROTOBUF(
          AcknowledgeOperationStatusMessage(),
          master.get()->pid,
          slave.get()->pid);
  
    // Resend the dropped operation status update.
    Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
    Clock::settle();
  
    AWAIT_READY(acknowledgeOperationStatusMessage);
  
    Clock::resume();
  }
  
  
  // This test verifies the master will adopt orphan operations reported
  // by agents upon reregistration. These orphans can appear whenever
  // the master fails over, but the operation's originating framework
  // never reregisters. Alternatively, and uncommonly, pending operations
  // on an agent migrated from one master to another will produce orphans.
  TEST_P(
      StorageLocalResourceProviderTest,
      TerminalOrphanOperationAfterMasterFailover)
  {
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath);
  
    setupResourceProviderConfig(Gigabytes(4));
  
    master::Flags masterFlags = CreateMasterFlags();
  
    // Greatly increase the ping timeout of the master/agent connection.
    // The clock will be advanced by `MIN_WAIT_BEFORE_ORPHAN_OPERATION_ADOPTION`
    // and if the ping timeout is less than this amount, the agent will
    // disconnect.
    masterFlags.agent_ping_timeout =
      master::DEFAULT_AGENT_PING_TIMEOUT +
      master::MIN_WAIT_BEFORE_ORPHAN_OPERATION_ADOPTION;
  
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    // NOTE: This test needs to use the StandaloneMasterDetector directly
    // (instead of the `cluster::Master::createDetector` helper) so that
    // the restarted master can be detected by the running agent, when the
    // test dictates.
    StandaloneMasterDetector detector(master.get()->pid);
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(&detector, slaveFlags);
    ASSERT_SOME(slave);
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // Register a framework to exercise an operation.
    v1::FrameworkInfo frameworkInfo = v1::DEFAULT_FRAMEWORK_INFO;
    frameworkInfo.set_roles(0, "storage");
  
    auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
  
    // NOTE: The scheduler may connect again after the TEARDOWN call.
    EXPECT_CALL(*scheduler, connected(_))
      .WillOnce(v1::scheduler::SendSubscribe(frameworkInfo))
      .WillRepeatedly(Return());
  
    Future<v1::scheduler::Event::Subscribed> subscribed;
    EXPECT_CALL(*scheduler, subscribed(_, _))
      .WillOnce(FutureArg<1>(&subscribed));
  
    EXPECT_CALL(*scheduler, heartbeat(_))
      .WillRepeatedly(Return()); // Ignore heartbeats.
  
    // Decline offers that contain only the agent's default resources.
    EXPECT_CALL(*scheduler, offers(_, _))
      .WillRepeatedly(v1::scheduler::DeclineOffers());
  
    // We are only interested in an offer with raw disk resources
    auto isRaw = [](const v1::Resource& r) {
      return r.has_disk() &&
        r.disk().has_source() &&
        r.disk().source().has_profile() &&
        r.disk().source().type() == v1::Resource::DiskInfo::Source::RAW;
    };
  
    Future<v1::scheduler::Event::Offers> offers;
  
    EXPECT_CALL(
        *scheduler,
        offers(_, v1::scheduler::OffersHaveAnyResource(isRaw)))
      .WillOnce(FutureArg<1>(&offers));
  
    v1::scheduler::TestMesos mesos(
        master.get()->pid,
        ContentType::PROTOBUF,
        scheduler);
  
    AWAIT_READY(subscribed);
  
    const v1::FrameworkID& frameworkId = subscribed->framework_id();
  
    // Prepare to intercept the terminal operation status message between
    // agent and master, to allow the master to failover first.
    // This creates a situation where the master forgets about the
    // framework that started the operation.
    Future<UpdateOperationStatusMessage> updateMessage =
      DROP_PROTOBUF(
          UpdateOperationStatusMessage(), slave.get()->pid, master.get()->pid);
  
    AWAIT_READY(offers);
    ASSERT_FALSE(offers->offers().empty());
  
    const v1::Offer& offer = offers->offers(0);
  
    ASSERT_FALSE(offer.resources().empty());
  
    // The test relies on retry logic within the SLRP module to resend the
    // terminal operation feedback message. We pause the clock here to control
    // when this retry happens.
    Clock::pause();
  
    // Have the framework call CREATE_DISK, a non-speculative operation.
    Option<v1::Resource> rawDisk;
  
    foreach (const v1::Resource& resource, offer.resources()) {
      if (isRaw(resource)) {
        rawDisk = resource;
        break;
      }
    }
  
    ASSERT_SOME(rawDisk);
  
    v1::OperationID operationId;
    operationId.set_value("operation");
  
    mesos.send(v1::createCallAccept(
        frameworkId,
        offer,
        {v1::CREATE_DISK(
            rawDisk.get(),
            v1::Resource::DiskInfo::Source::MOUNT,
            None(),
            operationId)}));
  
    // This message will be sent once the agent has completed the operation.
    AWAIT_READY(updateMessage);
  
    EXPECT_EQ(
        OperationState::OPERATION_FINISHED, updateMessage->status().state());
    EXPECT_EQ(
        operationId.value(), updateMessage->status().operation_id().value());
  
    Future<SlaveReregisteredMessage> slaveReregisteredMessage =
      FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
  
    // Trigger a master failover to enter a situation where a terminal operation
    // is orphaned, and the originating framework is not "completed".
    detector.appoint(None());
    master->reset();
  
    // Get rid of the framework so that it does not reregister.
    // We want the framework to remain unknown after the master/agent recovers.
    scheduler.reset();
  
    // Start the master back up and wait for the agent to reregister.
    master = StartMaster(masterFlags);
    ASSERT_SOME(master);
    detector.appoint(master.get()->pid);
  
    Clock::advance(slaveFlags.registration_backoff_factor);
    AWAIT_READY(slaveReregisteredMessage);
  
    Future<AcknowledgeOperationStatusMessage> acknowledgeOperationStatusMessage =
      FUTURE_PROTOBUF(
          AcknowledgeOperationStatusMessage(),
          master.get()->pid,
          slave.get()->pid);
  
    // Trigger a retry of the operation status update. The master should drop
    // this one because the agent has just reregistered and the framework is
    // unknown.
    Clock::advance(slave::STATUS_UPDATE_RETRY_INTERVAL_MIN);
    Clock::settle();
  
    ASSERT_TRUE(acknowledgeOperationStatusMessage.isPending());
  
    // Trigger a retry after advancing the clock beyond the operation "adoption"
    // time, which means the master should acknowledge the terminal status.
    Clock::advance(master::MIN_WAIT_BEFORE_ORPHAN_OPERATION_ADOPTION);
    Clock::settle();
  
    AWAIT_READY(acknowledgeOperationStatusMessage);
  
    Clock::resume();
  }
  
  
  // This test verifies that operators can reserve/unreserve and
  // create/destroy persistent volumes with resource provider resources.
  TEST_P(
      StorageLocalResourceProviderTest,
      OperatorOperationsWithResourceProviderResources)
  {
    Clock::pause();
  
    Future<shared_ptr<TestDiskProfileServer>> server =
      TestDiskProfileServer::create();
  
    AWAIT_READY(server);
  
    EXPECT_CALL(*server.get()->process, profiles(_))
      .WillOnce(Return(http::OK(createDiskProfileMapping({{"test", None()}}))))
      .WillRepeatedly(Return(Future<http::Response>())); // Stop subsequent polls.
  
    const Duration pollInterval = Seconds(10);
    loadUriDiskProfileAdaptorModule(
        stringify(server.get()->process->url()), pollInterval);
  
    setupResourceProviderConfig(Gigabytes(2));
  
    master::Flags masterFlags = CreateMasterFlags();
    Try<Owned<cluster::Master>> master = StartMaster(masterFlags);
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    // Since the local resource provider gets subscribed after the agent
    // is registered, it is guaranteed that the slave will send two
    // `UpdateSlaveMessage`s, where the latter one contains resources from
    // the storage local resource provider.
    //
    // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
    // Google Mock will search the expectations in reverse order.
    Future<UpdateSlaveMessage> updateSlave2 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
    Future<UpdateSlaveMessage> updateSlave1 =
      FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
  
    Future<SlaveRegisteredMessage> slaveRegisteredMessage =
      FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration and prevent retry.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    AWAIT_READY(updateSlave1);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(updateSlave2);
    ASSERT_TRUE(updateSlave2->has_resource_providers());
  
    Clock::pause();
  
    AWAIT_READY(slaveRegisteredMessage);
  
    // By default, all resource provider resources are reserved for the 'storage'
    // role. We'll refine this role to be able to add a reservation to the
    // existing one.
    const string role = string("storage/") + DEFAULT_TEST_ROLE;
  
    // Register a framework to create a 'MOUNT' volume and verify operator
    // operation results by checking received offers.
    FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
    framework.set_roles(0, role);
  
    MockScheduler sched;
    MesosSchedulerDriver driver(
        &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
  
    EXPECT_CALL(sched, registered(&driver, _, _));
  
    // Expect some offers to be rescinded as consequence of the operator
    // operations.
    EXPECT_CALL(sched, offerRescinded(_, _))
      .WillRepeatedly(Return());
  
    Future<vector<Offer>> offers;
  
    EXPECT_CALL(sched, resourceOffers(&driver, _))
      .WillRepeatedly(DeclineOffers());
  
    // Create a 'MOUNT' volume from 'RAW' resources, because persistent volumes
    // are only supported for 'MOUNT' volumes.
    {
      EXPECT_CALL(
          sched,
          resourceOffers(
              &driver,
              OffersHaveAnyResource(
                  std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
        .WillOnce(FutureArg<1>(&offers));
  
      driver.start();
  
      Clock::advance(masterFlags.allocation_interval);
  
      AWAIT_READY(offers);
      ASSERT_EQ(1u, offers->size());
  
      const Offer offer = offers->at(0);
  
      Resource raw = *Resources(offer.resources())
        .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
        .begin();
  
      EXPECT_CALL(
          sched,
          resourceOffers(
              &driver,
              OffersHaveAnyResource(
                  std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
        .WillOnce(FutureArg<1>(&offers));
  
      driver.acceptOffers(
          {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
  
      // NOTE: We need to resume the clock so that the resource provider can
      // periodically check if the CSI endpoint socket has been created by
      // the plugin container, which runs in another Linux process.
      Clock::resume();
  
      // Wait for an offer that contains the 'MOUNT' volume with default
      // reservation.
      AWAIT_READY(offers);
      ASSERT_EQ(1u, offers->size());
  
      Clock::pause();
    }
  
    Resource mountDisk = *Resources(offers->at(0).resources())
      .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
      .begin();
    const SlaveID slaveId = offers->at(0).slave_id();
  
    Resources reserved =
      Resources(mountDisk).pushReservation(createDynamicReservationInfo(
          role,
          DEFAULT_CREDENTIAL.principal()));
    reserved.unallocate();
  
    auto isReservedMountDisk = [](const Resource& r) {
      return
          r.has_disk() &&
          r.disk().has_source() &&
          r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
          r.disk().source().has_vendor() &&
          r.disk().source().vendor() == TEST_CSI_VENDOR &&
          r.disk().source().has_id() &&
          r.disk().source().has_profile() &&
          !r.disk().has_persistence() &&
          r.reservations().size() == 2; // Existing and refined reservation.
    };
  
    // Reserve resources as operator.
    {
      v1::master::Call v1ReserveResourcesCall;
      v1ReserveResourcesCall.set_type(v1::master::Call::RESERVE_RESOURCES);
  
      v1::master::Call::ReserveResources* reserveResources =
        v1ReserveResourcesCall.mutable_reserve_resources();
  
      reserveResources->mutable_agent_id()->CopyFrom(evolve(slaveId));
      reserveResources->mutable_resources()->CopyFrom(evolve(reserved));
  
      EXPECT_CALL(
          sched,
          resourceOffers(&driver, OffersHaveAnyResource(isReservedMountDisk)))
        .WillOnce(FutureArg<1>(&offers));
  
      Future<http::Response> v1ReserveResourcesResponse = http::post(
          master.get()->pid,
          "api/v1",
          createBasicAuthHeaders(DEFAULT_CREDENTIAL),
          serialize(ContentType::PROTOBUF, v1ReserveResourcesCall),
          stringify(ContentType::PROTOBUF));
  
      AWAIT_EXPECT_RESPONSE_STATUS_EQ(
          http::Accepted().status, v1ReserveResourcesResponse);
  
      Clock::advance(masterFlags.allocation_interval);
  
      // Wait for an offer that contains the 'MOUNT' volume with refined
      // reservation.
      AWAIT_READY(offers);
      ASSERT_EQ(1u, offers->size());
    }
  
    vector<Resource> converted;
    foreach (Resource resource, reserved) {
      if (Resources::isDisk(resource, Resource::DiskInfo::Source::MOUNT)) {
        resource.mutable_disk()->mutable_persistence()->set_id(
            id::UUID::random().toString());
        resource.mutable_disk()->mutable_persistence()->set_principal(
            DEFAULT_CREDENTIAL.principal());
        resource.mutable_disk()->mutable_volume()->set_container_path("volume");
        resource.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
        converted.push_back(resource);
      }
    }
  
    Resources volumes(converted);
  
    auto isPersistedMountDisk = [](const Resource& r) {
      return
          r.has_disk() &&
          r.disk().has_source() &&
          r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
          r.disk().source().has_vendor() &&
          r.disk().source().vendor() == TEST_CSI_VENDOR &&
          r.disk().source().has_id() &&
          r.disk().source().has_profile() &&
          r.disk().has_persistence();
    };
  
    // Create persistent volumes as operator.
    {
      v1::master::Call v1CreateVolumesCall;
      v1CreateVolumesCall.set_type(v1::master::Call::CREATE_VOLUMES);
  
      v1::master::Call::CreateVolumes* createVolumes =
        v1CreateVolumesCall.mutable_create_volumes();
  
      createVolumes->mutable_agent_id()->CopyFrom(evolve(slaveId));
      createVolumes->mutable_volumes()->CopyFrom(evolve(volumes));
  
      EXPECT_CALL(
          sched,
          resourceOffers(&driver, OffersHaveAnyResource(isPersistedMountDisk)))
        .WillOnce(FutureArg<1>(&offers));
  
      Future<http::Response> v1CreateVolumesResponse = http::post(
          master.get()->pid,
          "api/v1",
          createBasicAuthHeaders(DEFAULT_CREDENTIAL),
          serialize(ContentType::PROTOBUF, v1CreateVolumesCall),
          stringify(ContentType::PROTOBUF));
  
      AWAIT_EXPECT_RESPONSE_STATUS_EQ(
          http::Accepted().status, v1CreateVolumesResponse);
  
      // Remove all offer filters. Existing filters would stop us from
      // getting offers otherwise.
      driver.reviveOffers();
  
      Clock::advance(masterFlags.allocation_interval);
  
      // Wait for an offer that contains the 'MOUNT' volume with refined
      // reservation and persistence.
      AWAIT_READY(offers);
      ASSERT_EQ(1u, offers->size());
    }
  
    // Destroy persistent volumes as operator.
    {
      v1::master::Call v1DestroyVolumesCall;
      v1DestroyVolumesCall.set_type(v1::master::Call::DESTROY_VOLUMES);
  
      v1::master::Call::DestroyVolumes* destroyVolumes =
        v1DestroyVolumesCall.mutable_destroy_volumes();
  
      destroyVolumes->mutable_agent_id()->CopyFrom(evolve(slaveId));
      destroyVolumes->mutable_volumes()->CopyFrom(evolve(volumes));
  
      EXPECT_CALL(
          sched,
          resourceOffers(&driver, OffersHaveAnyResource(isReservedMountDisk)))
        .WillOnce(FutureArg<1>(&offers));
  
      Future<http::Response> v1DestroyVolumesResponse = http::post(
          master.get()->pid,
          "api/v1",
          createBasicAuthHeaders(DEFAULT_CREDENTIAL),
          serialize(ContentType::PROTOBUF, v1DestroyVolumesCall),
          stringify(ContentType::PROTOBUF));
  
      AWAIT_EXPECT_RESPONSE_STATUS_EQ(
          http::Accepted().status, v1DestroyVolumesResponse);
  
      Clock::advance(masterFlags.allocation_interval);
  
      // Wait for an offer that contains the 'MOUNT' volume with refined
      // reservation.
      AWAIT_READY(offers);
      ASSERT_EQ(1u, offers->size());
    }
  
    // Unreserve resources as operator.
    {
      v1::master::Call v1UnreserveResourcesCall;
      v1UnreserveResourcesCall.set_type(v1::master::Call::UNRESERVE_RESOURCES);
      v1::master::Call::UnreserveResources* unreserveResources =
        v1UnreserveResourcesCall.mutable_unreserve_resources();
  
      unreserveResources->mutable_agent_id()->CopyFrom(evolve(slaveId));
      unreserveResources->mutable_resources()->CopyFrom(evolve(reserved));
  
      EXPECT_CALL(
          sched,
          resourceOffers(
              &driver,
              OffersHaveAnyResource(
                  std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
        .WillOnce(FutureArg<1>(&offers));
  
      Future<http::Response> v1UnreserveResourcesResponse = http::post(
          master.get()->pid,
          "api/v1",
          createBasicAuthHeaders(DEFAULT_CREDENTIAL),
          serialize(ContentType::PROTOBUF, v1UnreserveResourcesCall),
          stringify(ContentType::PROTOBUF));
  
      AWAIT_EXPECT_RESPONSE_STATUS_EQ(
          http::Accepted().status, v1UnreserveResourcesResponse);
  
      Clock::advance(masterFlags.allocation_interval);
  
      // Wait for an offer that contains the 'MOUNT' volume with default
      // reservation.
      AWAIT_READY(offers);
      ASSERT_EQ(1u, offers->size());
    }
  }
  
  
  // This test validates that the SLRP periodically
  // reconciles resources with the CSI plugin.
  TEST_P(StorageLocalResourceProviderTest, Update)
  {
    Clock::pause();
  
    const string profilesPath = path::join(sandbox.get(), "profiles.json");
  
    ASSERT_SOME(
        os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
  
    loadUriDiskProfileAdaptorModule(profilesPath, Duration::max());
  
    const string mockCsiEndpoint =
      "unix://" + path::join(sandbox.get(), "mock_csi.sock");
  
    MockCSIPlugin plugin;
    ASSERT_SOME(plugin.startup(mockCsiEndpoint));
  
    constexpr Duration reconciliationInterval = Seconds(15);
  
    setupResourceProviderConfig(
        Bytes(0),
        None(),
        mockCsiEndpoint,
        None(),
        None(),
        reconciliationInterval);
  
    Try<Owned<cluster::Master>> master = StartMaster();
    ASSERT_SOME(master);
  
    Owned<MasterDetector> detector = master.get()->createDetector();
  
    process::Queue<Nothing> getCapacityCalls;
    process::Queue<Nothing> listVolumesCalls;
    if (GetParam() == csi::v0::API_VERSION) {
      EXPECT_CALL(plugin, ListVolumes(_, _, A<csi::v0::ListVolumesResponse*>()))
        .WillRepeatedly(Invoke([&](
            grpc::ServerContext* context,
            const csi::v0::ListVolumesRequest* request,
            csi::v0::ListVolumesResponse* response) {
          listVolumesCalls.put({});
          return grpc::Status::OK;
        }));
  
      EXPECT_CALL(plugin, GetCapacity(_, _, A<csi::v0::GetCapacityResponse*>()))
        .WillRepeatedly(Invoke([&](
            grpc::ServerContext* context,
            const csi::v0::GetCapacityRequest* request,
            csi::v0::GetCapacityResponse* response) {
          getCapacityCalls.put({});
          return grpc::Status::OK;
        }));
    } else if (GetParam() == csi::v1::API_VERSION) {
      EXPECT_CALL(plugin, ListVolumes(_, _, A<csi::v1::ListVolumesResponse*>()))
        .WillRepeatedly(Invoke([&](
            grpc::ServerContext* context,
            const csi::v1::ListVolumesRequest* request,
            csi::v1::ListVolumesResponse* response) {
          listVolumesCalls.put({});
          return grpc::Status::OK;
        }));
  
      EXPECT_CALL(plugin, GetCapacity(_, _, A<csi::v1::GetCapacityResponse*>()))
        .WillRepeatedly(Invoke([&](
            grpc::ServerContext* context,
            const csi::v1::GetCapacityRequest* request,
            csi::v1::GetCapacityResponse* response) {
          getCapacityCalls.put({});
          return grpc::Status::OK;
        }));
    }
  
    Future<Nothing> listVolumes1 = listVolumesCalls.get();
    Future<Nothing> listVolumes2 = listVolumesCalls.get();
  
    Future<Nothing> getCapacity1 = getCapacityCalls.get();
    Future<Nothing> getCapacity2 = getCapacityCalls.get();
  
    slave::Flags slaveFlags = CreateSlaveFlags();
    slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
  
    Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
    ASSERT_SOME(slave);
  
    // Advance the clock to trigger agent registration and prevent retry.
    Clock::advance(slaveFlags.registration_backoff_factor);
  
    // NOTE: We need to resume the clock so that the resource provider can
    // periodically check if the CSI endpoint socket has been created by
    // the plugin container, which runs in another Linux process.
    Clock::resume();
  
    AWAIT_READY(listVolumes1);
    AWAIT_READY(getCapacity1);
  
    Clock::pause();
  
    // Advance the clock so the SLRP polls for volume and storage pool updates.
    Clock::settle();
    Clock::advance(reconciliationInterval);
  
    AWAIT_READY(listVolumes2);
    AWAIT_READY(getCapacity2);
  }
  
  } // namespace tests {
  } // namespace internal {
  } // namespace mesos {
  