content: /*
   * 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.addservice;
  
  import static java.util.stream.Collectors.toList;
  import static java.util.stream.Collectors.toSet;
  import static org.apache.ambari.server.controller.internal.RequestResourceProvider.CONTEXT;
  
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Set;
  import java.util.stream.Stream;
  
  import jakarta.inject.Inject;
  import jakarta.inject.Singleton;
  
  import org.apache.ambari.server.AmbariException;
  import org.apache.ambari.server.controller.AmbariManagementController;
  import org.apache.ambari.server.controller.ClusterRequest;
  import org.apache.ambari.server.controller.ConfigurationRequest;
  import org.apache.ambari.server.controller.internal.ArtifactResourceProvider;
  import org.apache.ambari.server.controller.internal.ComponentResourceProvider;
  import org.apache.ambari.server.controller.internal.CredentialResourceProvider;
  import org.apache.ambari.server.controller.internal.HostComponentResourceProvider;
  import org.apache.ambari.server.controller.internal.RequestImpl;
  import org.apache.ambari.server.controller.internal.RequestOperationLevel;
  import org.apache.ambari.server.controller.internal.ServiceResourceProvider;
  import org.apache.ambari.server.controller.predicate.AndPredicate;
  import org.apache.ambari.server.controller.predicate.EqualsPredicate;
  import org.apache.ambari.server.controller.predicate.OrPredicate;
  import org.apache.ambari.server.controller.spi.ClusterController;
  import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
  import org.apache.ambari.server.controller.spi.NoSuchResourceException;
  import org.apache.ambari.server.controller.spi.Predicate;
  import org.apache.ambari.server.controller.spi.Request;
  import org.apache.ambari.server.controller.spi.Resource;
  import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
  import org.apache.ambari.server.controller.spi.ResourceProvider;
  import org.apache.ambari.server.controller.spi.SystemException;
  import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
  import org.apache.ambari.server.controller.utilities.ClusterControllerHelper;
  import org.apache.ambari.server.controller.utilities.PredicateBuilder;
  import org.apache.ambari.server.security.authorization.AuthorizationException;
  import org.apache.ambari.server.state.ConfigHelper;
  import org.apache.ambari.server.state.State;
  import org.apache.ambari.server.state.kerberos.KerberosDescriptor;
  import org.apache.ambari.server.state.kerberos.KerberosDescriptorFactory;
  import org.apache.ambari.server.topology.Credential;
  import org.apache.ambari.server.topology.ProvisionStep;
  import org.apache.ambari.server.utils.LoggingPreconditions;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.google.common.collect.ImmutableMap;
  import com.google.common.collect.ImmutableSet;
  import com.google.common.collect.Sets;
  
  /**
   * Creates resources using the resource providers.
   * Translates {@link AddServiceInfo} to internal requests accepted by those.
   */
  @Singleton
  public class ResourceProviderAdapter {
  
    private static final Logger LOG = LoggerFactory.getLogger(ResourceProviderAdapter.class);
    private static final LoggingPreconditions CHECK = new LoggingPreconditions(LOG);
  
    private final KerberosDescriptorFactory descriptorFactory = new KerberosDescriptorFactory();
  
    @Inject
    private AmbariManagementController controller;
  
    // business methods
  
    public void createServices(AddServiceInfo request) {
      LOG.info("Creating service resources for {}", request);
  
      Set<Map<String, Object>> properties = request.newServices().keySet().stream()
        .map(service -> createServiceRequestProperties(request, service))
        .collect(toSet());
  
      createResources(request, Resource.Type.Service, properties, null, false);
    }
  
    public void createComponents(AddServiceInfo request) {
      LOG.info("Creating component resources for {}", request);
  
      Set<Map<String, Object>> properties = request.newServices().entrySet().stream()
        .flatMap(componentsOfService -> componentsOfService.getValue().keySet().stream()
          .map(component -> createComponentRequestProperties(request, componentsOfService.getKey(), component)))
        .collect(toSet());
  
      createResources(request, Resource.Type.Component, properties, null, false);
    }
  
    public void createHostComponents(AddServiceInfo request) {
      LOG.info("Creating host component resources for {}", request);
  
      Set<Map<String, Object>> properties = request.newServices().entrySet().stream()
        .flatMap(componentsOfService -> componentsOfService.getValue().entrySet().stream()
          .flatMap(hostsOfComponent -> hostsOfComponent.getValue().stream()
            .map(host -> createHostComponentRequestProperties(request, componentsOfService.getKey(), hostsOfComponent.getKey(), host))))
        .collect(toSet());
  
      createResources(request, Resource.Type.HostComponent, properties, null, false);
    }
  
    public void createConfigs(AddServiceInfo request) {
      LOG.info("Creating configurations for {}", request);
      Set<ClusterRequest> requests = createConfigRequestsForNewServices(request);
      updateCluster(request, requests, "Error creating configurations for %s");
    }
  
    public void createCredentials(AddServiceInfo request) {
      if (!request.getRequest().getCredentials().isEmpty()) {
        LOG.info("Creating {} credential(s) for {}", request.getRequest().getCredentials().size(), request);
  
        request.getRequest().getCredentials().values().stream()
          .peek(credential -> LOG.debug("Creating credential {}", credential))
          .map(credential -> createCredentialRequestProperties(request.clusterName(), credential))
          .forEach(
            properties -> createResources(request, Resource.Type.Credential, ImmutableSet.of(properties), null, true)
          );
      }
    }
  
    /**
     * @return {@code Optional} with the cluster's kerberos descriptor artifact if it exists, otherwise empty {@code Optional}
     */
    public Optional<KerberosDescriptor> getKerberosDescriptor(AddServiceInfo request) {
      Set<String> propertyIds = ImmutableSet.of(ArtifactResourceProvider.ARTIFACT_DATA_PROPERTY);
      Predicate predicate = predicateForKerberosDescriptorArtifact(request.clusterName());
  
      Set<Resource> resources = getResources(request, propertyIds, Resource.Type.Artifact, predicate);
  
      if (resources == null || resources.isEmpty()) {
        return Optional.empty();
      }
  
      CHECK.checkArgument(resources.size() == 1,
        "Expected only one artifact of type %s, but got %d",
        ArtifactResourceProvider.KERBEROS_DESCRIPTOR,
        resources.size()
      );
  
      return Optional.of(descriptorFactory.createInstance(resources.iterator().next().getPropertiesMap().get(ArtifactResourceProvider.ARTIFACT_DATA_PROPERTY)));
    }
  
    public void createKerberosDescriptor(AddServiceInfo request, KerberosDescriptor descriptor) {
      LOG.info("Creating Kerberos descriptor for {}", request);
      Map<String, Object> properties = createKerberosDescriptorRequestProperties(request.clusterName());
      Map<String, String> requestInfo = requestInfoForKerberosDescriptor(descriptor);
      createResources(request, Resource.Type.Artifact, ImmutableSet.of(properties), requestInfo, false);
    }
  
    public void updateKerberosDescriptor(AddServiceInfo request, KerberosDescriptor descriptor) {
      LOG.info("Updating Kerberos descriptor from {}", request);
      Map<String, Object> properties = createKerberosDescriptorRequestProperties(request.clusterName());
      Map<String, String> requestInfo = requestInfoForKerberosDescriptor(descriptor);
      Predicate predicate = predicateForKerberosDescriptorArtifact(request.clusterName());
      updateResources(request, ImmutableSet.of(properties), Resource.Type.Artifact, predicate, requestInfo);
    }
  
    public void updateExistingConfigs(AddServiceInfo request, Set<String> existingServices) {
      LOG.info("Updating existing configurations for {}", request);
      Set<ClusterRequest> requests = createConfigRequestsForExistingServices(request, existingServices);
      updateCluster(request, requests, "Error updating configurations for %s");
    }
  
    public void updateServiceDesiredState(AddServiceInfo request, State desiredState) {
      LOG.info("Updating service desired state to {} for {}", desiredState, request);
  
      Set<Map<String, Object>> properties = ImmutableSet.of(ImmutableMap.of(
        ServiceResourceProvider.SERVICE_SERVICE_STATE_PROPERTY_ID, desiredState.name()
      ));
      Map<String, String> requestInfo = RequestOperationLevel.propertiesFor(Resource.Type.Service, request.clusterName());
      Predicate predicate = predicateForNewServices(request);
      updateResources(request, properties, Resource.Type.Service, predicate, requestInfo);
    }
  
    public void updateHostComponentDesiredState(AddServiceInfo request, Predicate predicate, ProvisionStep step) {
      State desiredState = step.getDesiredStateToSet();
      LOG.info("Updating host component desired state to {} per {} for {}", desiredState, step, request);
      LOG.debug("Using predicate {}", predicate);
  
      Set<Map<String, Object>> properties = ImmutableSet.of(ImmutableMap.of(
        HostComponentResourceProvider.STATE, desiredState.name(),
        CONTEXT, String.format("Put new components to %s state", desiredState)
      ));
  
      Map<String, String> requestInfo = new ImmutableMap.Builder<String, String>()
        .putAll(RequestOperationLevel.propertiesFor(Resource.Type.HostComponent, request.clusterName()))
        .putAll(step.getProvisionProperties())
        .build();
  
      HostComponentResourceProvider rp = (HostComponentResourceProvider) getClusterController().ensureResourceProvider(Resource.Type.HostComponent);
      Request internalRequest = createRequest(properties, requestInfo, null);
      try {
        rp.doUpdateResources(request.getStages(), internalRequest, predicate, false, false, false);
      } catch (UnsupportedPropertyException | SystemException | NoSuchParentResourceException | NoSuchResourceException e) {
        CHECK.wrapInUnchecked(e, RuntimeException::new, "Error updating host component desired state for %s", request);
      }
    }
  
    // ResourceProvider calls
  
    private static Set<Resource> getResources(AddServiceInfo request, Set<String> propertyIds, Resource.Type resourceType, Predicate predicate) {
      Request internalRequest = createRequest(null, null, propertyIds);
      ResourceProvider rp = getClusterController().ensureResourceProvider(resourceType);
      try {
        return rp.getResources(internalRequest, predicate);
      } catch (NoSuchResourceException e) {
        return ImmutableSet.of();
      } catch (UnsupportedPropertyException | SystemException | NoSuchParentResourceException e) {
        return CHECK.wrapInUnchecked(e, RuntimeException::new, "Error getting resources %s for %s", resourceType, request);
      }
    }
  
    private static void createResources(AddServiceInfo request, Resource.Type resourceType, Set<Map<String, Object>> properties, Map<String, String> requestInfo, boolean okIfExists) {
      Request internalRequest = createRequest(properties, requestInfo, null);
      ResourceProvider rp = getClusterController().ensureResourceProvider(resourceType);
      try {
        rp.createResources(internalRequest);
      } catch (UnsupportedPropertyException | SystemException | ResourceAlreadyExistsException | NoSuchParentResourceException e) {
        if (okIfExists && e instanceof ResourceAlreadyExistsException) {
          LOG.info("Resource already exists: {}, no need to create", e.getMessage());
        } else {
          CHECK.wrapInUnchecked(e, RuntimeException::new, "Error creating resources %s for %s", resourceType, request);
        }
      }
    }
  
    private static void updateResources(AddServiceInfo request, Set<Map<String, Object>> properties, Resource.Type resourceType, Predicate predicate, Map<String, String> requestInfo) {
      Request internalRequest = createRequest(properties, requestInfo, null);
      ResourceProvider rp = getClusterController().ensureResourceProvider(resourceType);
      try {
        rp.updateResources(internalRequest, predicate);
      } catch (UnsupportedPropertyException | SystemException | NoSuchParentResourceException | NoSuchResourceException e) {
        CHECK.wrapInUnchecked(e, RuntimeException::new, "Error updating resources %s for %s", resourceType, request);
      }
    }
  
    private void updateCluster(AddServiceInfo addServiceRequest, Set<ClusterRequest> requests, String errorMessageFormat) {
      try {
        controller.updateClusters(requests, null);
      } catch (AmbariException | AuthorizationException e) {
        CHECK.wrapInUnchecked(e, RuntimeException::new, errorMessageFormat, addServiceRequest);
      }
    }
  
    private static Request createRequest(Set<Map<String, Object>> properties, Map<String, String> requestInfoProperties, Set<String> propertyIds) {
      return new RequestImpl(propertyIds, properties, requestInfoProperties, null);
    }
  
    public static Map<String, String> requestInfoForKerberosDescriptor(KerberosDescriptor descriptor) {
      return ImmutableMap.of(Request.REQUEST_INFO_BODY_PROPERTY, ArtifactResourceProvider.toArtifactDataJson(descriptor.toMap()));
    }
  
    // request creation (as map of properties)
  
    private static Map<String, Object> createServiceRequestProperties(AddServiceInfo request, String service) {
      ImmutableMap.Builder<String, Object> properties = ImmutableMap.builder();
  
      properties.put(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID, request.clusterName());
      properties.put(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID, service);
      properties.put(ServiceResourceProvider.SERVICE_SERVICE_STATE_PROPERTY_ID, State.INIT.name());
  
      return properties.build();
    }
  
    private static Map<String, Object> createComponentRequestProperties(AddServiceInfo request, String service, String component) {
      ImmutableMap.Builder<String, Object> properties = ImmutableMap.builder();
  
      properties.put(ComponentResourceProvider.CLUSTER_NAME, request.clusterName());
      properties.put(ComponentResourceProvider.SERVICE_NAME, service);
      properties.put(ComponentResourceProvider.COMPONENT_NAME, component);
      properties.put(ComponentResourceProvider.STATE, State.INIT.name());
  
      return properties.build();
    }
  
    private static Map<String, Object> createHostComponentRequestProperties(AddServiceInfo request, String service, String component, String host) {
      ImmutableMap.Builder<String, Object> properties = ImmutableMap.builder();
  
      properties.put(HostComponentResourceProvider.CLUSTER_NAME, request.clusterName());
      properties.put(HostComponentResourceProvider.SERVICE_NAME, service);
      properties.put(HostComponentResourceProvider.COMPONENT_NAME, component);
      properties.put(HostComponentResourceProvider.HOST_NAME, host);
      properties.put(HostComponentResourceProvider.STATE, State.INIT.name());
  
      return properties.build();
    }
  
    public static Map<String, Object> createCredentialRequestProperties(String clusterName, Credential credential) {
      ImmutableMap.Builder<String, Object> properties = ImmutableMap.builder();
  
      properties.put(CredentialResourceProvider.CREDENTIAL_CLUSTER_NAME_PROPERTY_ID, clusterName);
      properties.put(CredentialResourceProvider.CREDENTIAL_ALIAS_PROPERTY_ID, credential.getAlias());
      properties.put(CredentialResourceProvider.CREDENTIAL_PRINCIPAL_PROPERTY_ID, credential.getPrincipal());
      properties.put(CredentialResourceProvider.CREDENTIAL_KEY_PROPERTY_ID, credential.getKey());
      properties.put(CredentialResourceProvider.CREDENTIAL_TYPE_PROPERTY_ID, credential.getType().name());
  
      return properties.build();
    }
  
    public static Map<String, Object> createKerberosDescriptorRequestProperties(String clusterName) {
      return ImmutableMap.of(
        ArtifactResourceProvider.CLUSTER_NAME_PROPERTY, clusterName,
        ArtifactResourceProvider.ARTIFACT_NAME_PROPERTY, ArtifactResourceProvider.KERBEROS_DESCRIPTOR
      );
    }
  
    // ClusterRequest creation (for configuration)
  
    private static Set<ClusterRequest> createConfigRequestsForNewServices(AddServiceInfo request) {
      Map<String, Map<String, String>> fullProperties = request.getConfig().getFullProperties();
      Map<String, Map<String, Map<String, String>>> fullAttributes = request.getConfig().getFullAttributes();
  
      return createConfigRequestsForServices(
        request.newServices().keySet(),
        configType -> !Objects.equals(configType, ConfigHelper.CLUSTER_ENV),
        request, fullProperties, fullAttributes
      );
    }
  
    private Set<ClusterRequest> createConfigRequestsForExistingServices(AddServiceInfo request, Set<String> existingServices) {
      Set<String> configTypesInRequest = ImmutableSet.copyOf(
        Sets.difference(
          Sets.union(
            request.getConfig().getProperties().keySet(),
            request.getConfig().getAttributes().keySet()),
          ImmutableSet.of(ConfigHelper.CLUSTER_ENV))
      );
  
      Map<String, Map<String, String>> fullProperties = request.getConfig().getFullProperties();
      Map<String, Map<String, Map<String, String>>> fullAttributes = request.getConfig().getFullAttributes();
  
      Set<ClusterRequest> clusterRequests = createConfigRequestsForServices(
        existingServices,
        configTypesInRequest::contains,
        request, fullProperties, fullAttributes
      );
  
      if (request.getConfig().getProperties().containsKey(ConfigHelper.CLUSTER_ENV)) {
        Optional<ClusterRequest> clusterEnvRequest = createConfigRequestForConfigTypes(Stream.of(ConfigHelper.CLUSTER_ENV),
          request, fullProperties, fullAttributes);
        clusterEnvRequest.ifPresent(clusterRequests::add);
      }
  
      return clusterRequests;
    }
  
    private static Set<ClusterRequest> createConfigRequestsForServices(
      Set<String> services,
      java.util.function.Predicate<String> predicate,
      AddServiceInfo request,
      Map<String, Map<String, String>> fullProperties,
      Map<String, Map<String, Map<String, String>>> fullAttributes
    ) {
      return services.stream()
        .map(service -> createConfigRequestForConfigTypes(
          request.getStack().getConfigurationTypes(service).stream()
            .filter(predicate),
          request, fullProperties, fullAttributes
        ))
        .filter(Optional::isPresent)
        .map(Optional::get)
        .collect(toSet());
    }
  
    /**
     * Creates a {@link ConfigurationRequest} for each config type in the {@code configTypes} stream.
     *
     * @return an {@code Optional} {@link ClusterRequest} with desired configs set to all the {@code ConfigurationRequests},
     * or an empty {@code Optional} if the incoming {@code configTypes} stream is empty
     */
    private static Optional<ClusterRequest> createConfigRequestForConfigTypes(
      Stream<String> configTypes,
      AddServiceInfo addServiceRequest,
      Map<String, Map<String, String>> fullProperties,
      Map<String, Map<String, Map<String, String>>> fullAttributes
    ) {
      List<ConfigurationRequest> configRequests = configTypes
        .peek(configType -> LOG.info("Creating request for config type {} for {}", configType, addServiceRequest))
        .map(configType -> new ConfigurationRequest(addServiceRequest.clusterName(), configType, "ADD_SERVICE_" + System.currentTimeMillis(),
          fullProperties.getOrDefault(configType, ImmutableMap.of()),
          fullAttributes.getOrDefault(configType, ImmutableMap.of())))
        .collect(toList());
  
      if (configRequests.isEmpty()) {
        return Optional.empty();
      }
  
      ClusterRequest clusterRequest = new ClusterRequest(null, addServiceRequest.clusterName(), null, null);
      clusterRequest.setDesiredConfig(configRequests);
      return Optional.of(clusterRequest);
    }
  
    // Predicate creation
  
    public static Predicate predicateForKerberosDescriptorArtifact(String clusterName) {
      return new PredicateBuilder().begin()
        .property(ArtifactResourceProvider.CLUSTER_NAME_PROPERTY).equals(clusterName)
        .and()
        .property(ArtifactResourceProvider.ARTIFACT_NAME_PROPERTY).equals(ArtifactResourceProvider.KERBEROS_DESCRIPTOR)
        .end().toPredicate();
    }
  
    private static Predicate predicateForNewServices(AddServiceInfo request) {
      return new AndPredicate(
        new EqualsPredicate<>(ServiceResourceProvider.SERVICE_CLUSTER_NAME_PROPERTY_ID, request.clusterName()),
        OrPredicate.of(
          request.newServices().keySet().stream()
            .map(service -> new EqualsPredicate<>(ServiceResourceProvider.SERVICE_SERVICE_NAME_PROPERTY_ID, service))
            .collect(toList())
        )
      );
    }
  
    // TODO should be injected
    private static ClusterController getClusterController() {
      return ClusterControllerHelper.getClusterController();
    }
  
  }
  