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 "resource_provider/storage/disk_profile_utils.hpp"
  
  #include <google/protobuf/util/json_util.h>
  
  #include <stout/bytes.hpp>
  #include <stout/check.hpp>
  #include <stout/foreach.hpp>
  #include <stout/unreachable.hpp>
  
  using std::string;
  
  using mesos::resource_provider::DiskProfileMapping;
  
  namespace mesos {
  namespace internal {
  namespace storage {
  
  Try<DiskProfileMapping> parseDiskProfileMapping(
      const string& data)
  {
    // Use Google's JSON utility function to parse the JSON string.
    DiskProfileMapping output;
    google::protobuf::util::JsonParseOptions options;
    options.ignore_unknown_fields = true;
  
    google::protobuf::util::Status status =
      google::protobuf::util::JsonStringToMessage(data, &output, options);
  
    if (!status.ok()) {
      return Error(
          "Failed to parse DiskProfileMapping message: " +
          status.ToString());
    }
  
    Option<Error> validation = validate(output);
    if (validation.isSome()) {
      return Error(
          "Fetched profile mapping failed validation with: " +
          validation->message);
    }
  
    return output;
  }
  
  
  bool isSelectedResourceProvider(
      const DiskProfileMapping::CSIManifest& profileManifest,
      const ResourceProviderInfo& resourceProviderInfo)
  {
    switch (profileManifest.selector_case()) {
      case DiskProfileMapping::CSIManifest::kResourceProviderSelector: {
        CHECK(profileManifest.has_resource_provider_selector());
  
        const auto& selector = profileManifest.resource_provider_selector();
  
        foreach (const auto& resourceProvider, selector.resource_providers()) {
          if (resourceProviderInfo.type() == resourceProvider.type() &&
              resourceProviderInfo.name() == resourceProvider.name()) {
            return true;
          }
        }
  
        return false;
      }
      case DiskProfileMapping::CSIManifest::kCsiPluginTypeSelector: {
        CHECK(profileManifest.has_csi_plugin_type_selector());
  
        const auto& selector = profileManifest.csi_plugin_type_selector();
  
        return resourceProviderInfo.has_storage() &&
          resourceProviderInfo.storage().plugin().type() ==
            selector.plugin_type();
      }
      case DiskProfileMapping::CSIManifest::SELECTOR_NOT_SET: {
        UNREACHABLE();
      }
    }
  
    UNREACHABLE();
  }
  
  
  static Option<Error> validateSelector(
      const DiskProfileMapping::CSIManifest& profileManifest)
  {
    switch (profileManifest.selector_case()) {
      case DiskProfileMapping::CSIManifest::kResourceProviderSelector: {
        if (!profileManifest.has_resource_provider_selector()) {
          return Error("Expecting 'resource_provider_selector' to be present");
        }
  
        const auto& selector = profileManifest.resource_provider_selector();
  
        foreach (const auto& resourceProvider, selector.resource_providers()) {
          if (resourceProvider.type().empty()) {
            return Error(
                "'type' is a required field for ResourceProviderSelector");
          }
  
          if (resourceProvider.name().empty()) {
            return Error(
                "'name' is a required field for ResourceProviderSelector");
          }
        }
  
        break;
      }
      case DiskProfileMapping::CSIManifest::kCsiPluginTypeSelector: {
        if (!profileManifest.has_csi_plugin_type_selector()) {
          return Error("Expecting 'csi_plugin_type_selector' to be present");
        }
  
        const auto& selector = profileManifest.csi_plugin_type_selector();
  
        if (selector.plugin_type().empty()) {
          return Error(
              "'plugin_type' is a required field for CSIPluginTypeSelector");
        }
  
        break;
      }
      case DiskProfileMapping::CSIManifest::SELECTOR_NOT_SET: {
        return Error("Expecting a selector to be present");
      }
    }
  
    return None();
  }
  
  
  Option<Error> validate(const DiskProfileMapping& mapping)
  {
    foreach (const auto& entry, mapping.profile_matrix()) {
      Option<Error> selector = validateSelector(entry.second);
  
      if (selector.isSome()) {
        return Error(
            "Profile '" + entry.first + "' has no valid selector: " +
            selector->message);
      }
  
      if (!entry.second.has_volume_capabilities()) {
        return Error(
            "Profile '" + entry.first +
            "' is missing the required field 'volume_capabilities'");
      }
  
      Option<Error> capabilities = validate(entry.second.volume_capabilities());
  
      if (capabilities.isSome()) {
        return Error(
            "Profile '" + entry.first + "' has an invalid VolumeCapability: " +
            capabilities->message);
      }
  
      // NOTE: The `create_parameters` field is optional and needs no further
      // validation after parsing.
    }
  
    return None();
  }
  
  
  // TODO(chhsiao): Move this to CSI validation implementation file.
  Option<Error> validate(const csi::v0::VolumeCapability& capability)
  {
    if (capability.has_mount()) {
      // The total size of this repeated field may not exceed 4 KB.
      //
      // TODO(josephw): The specification does not state how this maximum
      // size is calculated. So this check is conservative and does not
      // include padding or array separators in the size calculation.
      size_t size = 0;
      foreach (const string& flag, capability.mount().mount_flags()) {
        size += flag.size();
      }
  
      if (Bytes(size) > Kilobytes(4)) {
        return Error("Size of 'mount_flags' may not exceed 4 KB");
      }
    }
  
    if (!capability.has_access_mode()) {
      return Error("'access_mode' is a required field");
    }
  
    if (capability.access_mode().mode() ==
        csi::v0::VolumeCapability::AccessMode::UNKNOWN) {
      return Error("'access_mode.mode' is unknown or not set");
    }
  
    return None();
  }
  
  } // namespace storage {
  } // namespace internal {
  } // namespace mesos {
  