ambari-server/src/main/java/org/apache/ambari/server/topology/ConfigurableHelper.java

/*
 * 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.
 */

package org.apache.ambari.server.topology;

import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.ambari.server.controller.internal.BlueprintResourceProvider.PROPERTIES_ATTRIBUTES_PROPERTY_ID;
import static org.apache.ambari.server.controller.internal.BlueprintResourceProvider.PROPERTIES_PROPERTY_ID;
import static org.apache.ambari.server.topology.ConfigurationFactory.isKeyInLegacyFormat;
import static org.apache.ambari.server.topology.ConfigurationFactory.isKeyInNewFormat;
import static org.apache.ambari.server.topology.ConfigurationFactory.splitConfigurationKey;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

import javax.annotation.Nullable;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;

/**
 * Helper class for {@link Configurable}
 */
public class ConfigurableHelper {

  private static final ImmutableSet<String> PERMITTED_CONFIG_FIELDS = ImmutableSet.of(PROPERTIES_PROPERTY_ID, PROPERTIES_ATTRIBUTES_PROPERTY_ID);

  /**
   * Parses configuration maps The configs can be in fully structured JSON, e.g.
   * <code>
   * [{"hdfs-site":
   *  "properties": {
   *    ""dfs.replication": "3",
   *    ...
   *  },
   *  properties_attributes: {}
   * }]
   * </code>
   * or flattened like
   * <code>
   * [{
   *  "hdfs-site/properties/dfs.replication": "3",
   *  ...
   * }]
   * </code>
   * In the latter case it calls {@link ConfigurationFactory#getConfiguration(Collection)}
   */
  public static Configuration parseConfigs(@Nullable Collection<? extends Map<String, ?>> configs) {
    Configuration configuration;

    if (null == configs || configs.isEmpty()) {
      configuration = Configuration.newEmpty();
    }
    else if (configs.iterator().next().keySet().iterator().next().contains("/")) {
      // Configuration has keys with slashes like "zk.cfg/properties/dataDir" means it is coming through
      // the resource framework and must be parsed with configuration factories
      checkFlattenedConfig(configs);
      configuration = new ConfigurationFactory().getConfiguration((Collection<Map<String, String>>)configs);
    }
    else {
      // If the configuration does not have keys with slashes it means it is coming from plain JSON and needs to be
      // parsed accordingly.
      Map<String, Map<String, String>> allProperties = new HashMap<>();
      Map<String, Map<String, Map<String, String>>> allAttributes = new HashMap<>();
      configs.forEach( item -> {

        checkArgument(item.size() == 1, "Each config object must have a single property which is the name of the config," +
          " e.g. \"cluster-env\" : {...}");

        Map.Entry<String, ?> configEntry = item.entrySet().iterator().next();
        String configName = item.keySet().iterator().next();

        checkArgument(configEntry.getValue() instanceof Map,
          "The value for %s must be a JSON object (found: %s)", configName, getClassName(configEntry.getValue()));

        Map<String, ?> configData = (Map<String, ?>) configEntry.getValue();

        Set<String> extraKeys = Sets.difference(configData.keySet(), PERMITTED_CONFIG_FIELDS);
        boolean legacy = extraKeys.size() == configData.keySet().size();
        checkArgument(legacy || extraKeys.isEmpty(), "Invalid fields in %s configuration: %s", configName, extraKeys);

        if (legacy) {
          checkMap("don't care", configData, String.class);
          Map<String, String> properties = (Map<String, String>)configData;
          allProperties.put(configName, properties);
        } else {
          if (configData.containsKey(PROPERTIES_PROPERTY_ID)) {
            checkMap(PROPERTIES_PROPERTY_ID, configData.get(PROPERTIES_PROPERTY_ID), String.class);

            Map<String, String> properties = (Map<String, String>) configData.get(PROPERTIES_PROPERTY_ID);
            allProperties.put(configName, properties);
          }
          if (configData.containsKey(PROPERTIES_ATTRIBUTES_PROPERTY_ID)) {
            checkMap(PROPERTIES_ATTRIBUTES_PROPERTY_ID, configData.get(PROPERTIES_ATTRIBUTES_PROPERTY_ID), Map.class);
            Map<String, Map<String, String>> attributes =
              (Map<String, Map<String, String>>) configData.get(PROPERTIES_ATTRIBUTES_PROPERTY_ID);
            attributes.forEach((key, value) -> checkMap(key, value, String.class));

            allAttributes.put(configName, attributes);
          }
        }
      });

      configuration = new Configuration(allProperties, allAttributes);
    }
    return configuration;
  }

  /**
   * Converts {@link Configuration} objects to a collection easily serializable to Json
   * @param configuration the configuration to convert
   * @return the resulting collection
   */
  public static Collection<Map<String, Map<String, ?>>> convertConfigToMap(Configuration configuration) {
    Collection<Map<String, Map<String, ?>>> configurations = new ArrayList<>();
    Set<String> allConfigTypes = Sets.union(configuration.getProperties().keySet(), configuration.getAttributes().keySet());
    for (String configType: allConfigTypes) {
      Map<String, Map<String, ?>> configData = new HashMap<>();
      if (configuration.getProperties().containsKey(configType)) {
        configData.put(PROPERTIES_PROPERTY_ID, configuration.getProperties().get(configType));
      }
      if (configuration.getAttributes().containsKey(configType)) {
        configData.put(PROPERTIES_ATTRIBUTES_PROPERTY_ID, configuration.getAttributes().get(configType));
      }
      configurations.add(ImmutableMap.of(configType, configData));
    }
    return configurations;
  }

  private static void checkFlattenedConfig(Collection<? extends Map<String, ?>> configs) {
    configs.forEach( config -> {
      if ( !config.isEmpty() ) {
        List<String> firstKey = splitConfigurationKey(config.keySet().iterator().next());
        String configType = firstKey.get(0);
        boolean legacyConfig = isKeyInLegacyFormat(firstKey);

        config.keySet().forEach( key -> {
          List<String> keyParts = splitConfigurationKey(key);
          checkArgument(Objects.equals(configType, keyParts.get(0)),
            "Invalid config type: %s. Should be: %s", keyParts.get(0), configType);
          checkArgument(legacyConfig && isKeyInLegacyFormat(keyParts) || !legacyConfig && isKeyInNewFormat(keyParts),
            "Expected key in %s format, found [%s]",
            legacyConfig ? "[config_type/property_name]" :
              "[config_type/properties/config_name] or [config_type/properties_attributes/attribute_name/property_name]",
            key);
        });
      }
    });
  }

  private static void checkMap(String fieldName, Object mapObj, Class<?> valueType) {
    checkArgument(mapObj instanceof Map, "'%s' needs to be a JSON object. Found: %s", fieldName, getClassName(mapObj));
    Map<?, ?> map = (Map<?, ?>)mapObj;
    map.forEach( (__, value) ->
      checkArgument(valueType.isInstance(value),
        "Expected %s as value type, found %s, type: %s",
        valueType.getName(), value, getClassName(value)));
  }

  private static String getClassName(Object object) {
    return null != object ? object.getClass().getName() : null;
  }

}