content: package alien4cloud.dao;
  
  import java.beans.IntrospectionException;
  import java.io.IOException;
  import java.lang.reflect.Field;
  import java.lang.reflect.Modifier;
  import java.util.Collection;
  import java.util.ArrayList;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  
  import javax.annotation.PreDestroy;
  import javax.annotation.Resource;
  
  import org.apache.commons.lang3.StringUtils;
  import org.apache.lucene.util.NamedThreadFactory;
  import org.elasticsearch.action.ActionFuture;
  import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
  import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
  import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
  import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
  import org.elasticsearch.client.Client;
  import org.elasticsearch.common.settings.Settings;
  import org.elasticsearch.common.xcontent.XContentType;
  import org.elasticsearch.index.query.QueryBuilders;
  import org.elasticsearch.index.reindex.BulkByScrollResponse;
  import org.elasticsearch.index.reindex.DeleteByQueryAction;
  import org.elasticsearch.index.reindex.DeleteByQueryRequestBuilder;
  import org.elasticsearch.mapping.ElasticSearchClient;
  import org.elasticsearch.mapping.FieldsMappingBuilder;;
  import org.elasticsearch.mapping.MappingBuilder;
  import org.elasticsearch.util.MapUtil;
  
  import org.elasticsearch.annotation.ESObject;
  
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.google.common.collect.Lists;
  import com.google.common.collect.Maps;
  
  import org.springframework.beans.factory.annotation.Value;
  
  import org.elasticsearch.annotation.Id;
  import alien4cloud.exception.IndexingServiceException;
  import alien4cloud.rest.utils.JsonUtil;
  import alien4cloud.utils.ReflectionUtil;
  import lombok.Getter;
  import lombok.Setter;
  import lombok.SneakyThrows;
  import lombok.extern.slf4j.Slf4j;
  
  /**
   * Manages one or multiple indexes and the related java and elastic search types.
   *
   * @author luc boutier
   */
  @Slf4j
  public abstract class ESIndexMapper {
      /** Custom score field for alien. */
      public static final String ALIEN_SCORE = "alienScore";
  
      /** Unique type name for indices, as recommended by ES */
      public static final String TYPE_NAME = "_doc";
  
      @Resource
      private ElasticSearchClient esClient;
      @Resource
      @Getter
      private MappingBuilder mappingBuilder;
  
      @Getter
      private final Map<String, String> typesToIndices = Maps.newHashMap();
      @Getter
      private final Map<String, Class<?>> typesToClasses = Maps.newHashMap();
  
      private final Map<String, List<String>> classesToIndicesGroups = Maps.newHashMap();
      public List<String> getClassesToIndicesGroups(String clazz) {
        return classesToIndicesGroups.get(clazz);
      }
  
      /** If a class has an id field generated by elasticsearch then the field is specified here. */
      @Getter
      private final Map<Class<?>, Field> classTogeneratedIdFields = Maps.newHashMap();
  
      @Getter
      private String[] allIndexes;
      @Getter
      @Setter
      private ObjectMapper jsonMapper = new ObjectMapper();
  
      /** TTL scanning period, default is 1 day */
      @Value("${ttl.period:86400}")
      private String ttlPeriod;
  
      @Value("${elasticSearch.shard_count:-1}")
      private Integer shardCount;
  
      @Value("${elasticSearch.replica_count:-1}")
      private Integer replicaCount;
  
      @Value("${elasticSearch._initIndex:true}")
      private Boolean initIndex;
  
      @Value("${elasticSearch._nowait:false}")
      private Boolean nowait;
  
      private List<TTL> TTLs = new ArrayList<TTL>();
  
      private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1, new NamedThreadFactory("es-ttl-cleaner"));
  
      /**
       * Initialize the array of all indices managed by this dao.
       */
      public void initCompleted() {
          Set<String> indices = typesToClasses.entrySet().stream()
                  .filter(e -> ! Modifier.isAbstract(e.getValue().getModifiers()) && e.getValue().isAnnotationPresent(ESObject.class))
                  .map( e -> e.getKey())
                  .collect(Collectors.toSet());
  
          allIndexes = indices.toArray(new String[indices.size()]);
  
          if (!nowait) {
              esClient.waitForGreenStatus(allIndexes);
          }
          if ((ttlPeriod != null) &&
              !TTLs.isEmpty()) {
             long period = Long.valueOf(ttlPeriod).longValue();
             if (period > 0) {
                executorService.scheduleWithFixedDelay(new Runnable() {
                   @Override
                   public void run() {
                      log.debug ("Scanning TTLs");
                      for (TTL ttl : TTLs) {
                         cleanIndex (ttl.index, ttl.field, ttl.ttl);
                      }
                   }
                }, period, period, TimeUnit.SECONDS);
             }
          }
      }
  
      @PreDestroy
      public void destroy() {
          executorService.shutdown();
          try {
              executorService.awaitTermination(5, TimeUnit.MINUTES);
          } catch (InterruptedException e) {
          }
      }
  
      /**
       * Create if not exist indices.
       * A TTL can be defined for all indices under this index (ESearch TTL notation)
       *
       * @param indexName The index to initialize
       * @param classes   An array of classes to map to this index.
       */
      @SneakyThrows({IOException.class, IntrospectionException.class})
      public void initIndices(String indexName, String ttl, Class<?>... classes) {
  
          for (Class<?> clazz : classes) {
              String typeName = addToMappedClasses(indexName, clazz);
  
              if ((ttl != null) &&
                      !Modifier.isAbstract(clazz.getModifiers()) &&
                      clazz.isAnnotationPresent(ESObject.class)) {
                  String ts = (new FieldsMappingBuilder()).getTimeStamp(clazz);
                  if (ts == null) {
                      ts = "timestamp";
                  }
                  TTLs.add(new TTL(typeName, ts, ttl));
              }
  
              if (indexExist(typeName)) {
                  Field generatedIdField = ReflectionUtil.getDeclaredField(clazz, EsGeneratedId.class);
                  if (generatedIdField != null) {
                      generatedIdField.setAccessible(true);
                      classTogeneratedIdFields.put(clazz, generatedIdField);
                  }
                  generatedIdField = ReflectionUtil.getDeclaredField(clazz, Id.class);
                  if (generatedIdField != null) {
                      generatedIdField.setAccessible(true);
                      classTogeneratedIdFields.put(clazz, generatedIdField);
                  }
  
                  continue;
              }
              if (Modifier.isAbstract(clazz.getModifiers()) ||
                      !clazz.isAnnotationPresent(ESObject.class)) {
                  continue; // no mapping to register for abstract classes.
              }
  
              CreateIndexRequestBuilder createIndexRequestBuilder = esClient.getClient().admin().indices().prepareCreate(typeName);
              String typeMapping = mappingBuilder.getMapping(clazz);
              Map<String, Object> typesMap = JsonUtil.toMap(typeMapping);
  
              addAlienScore(typesMap);
              //addTTL(typesMap, ttl); not supported in ES 5
  
              Field generatedIdField = ReflectionUtil.getDeclaredField(clazz, EsGeneratedId.class);
              if (generatedIdField != null) {
                  generatedIdField.setAccessible(true);
                  classTogeneratedIdFields.put(clazz, generatedIdField);
              }
              generatedIdField = ReflectionUtil.getDeclaredField(clazz, Id.class);
              if (generatedIdField != null) {
                  generatedIdField.setAccessible(true);
                  classTogeneratedIdFields.put(clazz, generatedIdField);
              }
  
              String mapping = jsonMapper.writeValueAsString(typesMap);
              //createIndexRequestBuilder.addMapping(typeName, mapping, XContentType.JSON);
              createIndexRequestBuilder.addMapping(TYPE_NAME, mapping, XContentType.JSON);
  
              Settings.Builder settingsBuilder = Settings.builder();
  
              // add settings if any (including analysers definitions)
              String indexSettings = mappingBuilder.getIndexSettings(clazz);
              if (StringUtils.isNotBlank(indexSettings)) {
                  settingsBuilder.loadFromSource(indexSettings, XContentType.JSON);
                  //createIndexRequestBuilder.setSettings(indexSettings, XContentType.JSON);
              }
  
              if (shardCount > 0) {
                  settingsBuilder.put("index.number_of_shards", shardCount);
              }
  
              if (replicaCount >= 0) {
                  settingsBuilder.put("index.number_of_replicas", replicaCount);
              }
  
              createIndexRequestBuilder.setSettings(settingsBuilder);
  
              if (!initIndex) {
                  log.warn("Init index is disabled !");
              } else {
                  try {
                      final CreateIndexResponse createResponse = createIndexRequestBuilder.execute().actionGet();
                      if (!createResponse.isAcknowledged()) {
                          throw new IndexingServiceException("Failed to create index <" + indexName + ">");
                      }
                  } catch (Exception e) {
                      log.warn("Not able to init indice for index {}, maybe it has been created elsewhere", indexName);
                      log.error(e.getMessage());
                  }
              }
          }
      }
  
      /**
       * Add the alien score field for each type in the map.
       * 
       * @param typesMap The type map.
       */
      public static void addAlienScore(Map<String, Object> typesMap) {
          for (Object typeMappingObject : typesMap.values()) {
              @SuppressWarnings("unchecked")
              Map<String, Object> typeMappingMap = (Map<String, Object>) typeMappingObject;
              Map<String, Object> propertiesMap = (Map<String, Object>) typeMappingMap.get("properties");
              Map<String, Object> scoreMapping = MapUtil.getMap(
                      new String[] { "index", "boost", "store", "ignore_malformed", "type" },
                      new String[] { "true", "1.0", "false", "false", "long" });
              propertiesMap.put(ALIEN_SCORE, scoreMapping);
          }
      }
  
      private void cleanIndex (String index, String ts, String ttl) {
         log.info ("Cleaning index " + index + "(" + ts + ") ttl=" + ttl);
         try {
             BulkByScrollResponse response = new DeleteByQueryRequestBuilder(esClient.getClient(), DeleteByQueryAction.INSTANCE)
                     .source(index)
                     .filter(QueryBuilders.rangeQuery(ts).lte("now-" + ttl))
                     .get();
             log.info ("Deleted " + response.getDeleted() + " docs");
         } catch(Exception e) {
            log.error("An error occured while cleaning (TTL) index " + index, e);
         }
  
      }
  
      @SneakyThrows({ ExecutionException.class, InterruptedException.class })
      protected boolean indexExist(String indexName) {
          // check if existing before
          final ActionFuture<IndicesExistsResponse> indexExistFuture = esClient.getClient().admin().indices().exists(new IndicesExistsRequest(indexName));
          IndicesExistsResponse response;
          response = indexExistFuture.get();
          return response.isExists();
      }
  
  /******************
      private void addToMappedClasses(String indexName, Class<?>[] classes) {
          for (Class<?> clazz : classes) {
              addToMappedClasses(indexName, clazz);
              Field generatedIdField = ReflectionUtil.getDeclaredField(clazz, EsGeneratedId.class);
              if (generatedIdField != null) {
                  generatedIdField.setAccessible(true);
                  classTogeneratedIdFields.put(clazz, generatedIdField);
              }
              generatedIdField = ReflectionUtil.getDeclaredField(clazz, Id.class);
              if (generatedIdField != null) {
                  generatedIdField.setAccessible(true);
                  classTogeneratedIdFields.put(clazz, generatedIdField);
              }
          }
      }
  ************************/
  
      private String addToMappedClasses(String indexName, Class<?> clazz) {
          //log.info("Mapping class <" + clazz.getName() + "> to index <" + indexName + ">");
          String typeName = MappingBuilder.indexTypeFromClass(clazz);
          typesToIndices.put(typeName, indexName);
          typesToClasses.put(typeName, clazz);
  
          if (!Modifier.isAbstract(clazz.getModifiers()) &&
              clazz.isAnnotationPresent(ESObject.class)) {
             List<String> indices = classesToIndicesGroups.get(indexName);
             if (indices == null) {
                indices = new ArrayList<String>();
                classesToIndicesGroups.put (indexName, indices);
             }
             indices.add(typeName);
          }
          log.info("Mapping class <" + clazz.getName() + "> to index <" + typeName + ">");
  
          return typeName;
      }
  
      /**
       * Get the index in which the given type lies.
       *
       * @param clazz The type for which to get the index.
       * @return The index in which the given type lies.
       */
      public String[] getIndexForType(Class<?> clazz) {
          String typeName = MappingBuilder.indexTypeFromClass(clazz);
          String index = typesToIndices.get(typeName);
          if ((index == null) || (classesToIndicesGroups.get(index) == null)) {
              log.error("Class <" + clazz.getName() + "> is not registered in any indexes.");
              throw new IndexingServiceException("Requested type <" + typeName + "> is not registered in any indexes.");
          }
          if (Modifier.isAbstract(clazz.getModifiers()) || // abstract class : get all associated indices
              !clazz.isAnnotationPresent(ESObject.class)) {
             List<String> lresult = classesToIndicesGroups.get(index);
             String[] tresult = new String[lresult.size()];
             return lresult.toArray(tresult);
          } else { // concrete class index is class in lower case
             return new String[] { typeName };
         }
      }
  
      /**
       * Return a class from the given elastic search type.
       *
       * @param type The elastic search type.
       * @return The class matching the given type if any, null if no class is matching the type.
       */
      public Class<?> getClassFromType(String type) {
          return this.typesToClasses.get(type);
      }
  
      /**
       * Get the class from the class name.
       *
       * @param className The name of the class to get.
       * @param allowNull If a null class name is allowed. If className is null and allowNull is true then null is returned, if className is null and allowNull is
       *            false then an {@link IndexingServiceException} is thrown.
       * @return The class matching the given class name, or null if className is null and allowNull is true.
       */
      public Class<?> getClassFromName(String className, boolean allowNull) {
          if (className == null || className.trim().isEmpty()) {
              if (allowNull) {
                  return null;
              }
              throw new IndexingServiceException("Class type <" + className + "> not found.");
          } else {
              try {
                  return Class.forName(className);
              } catch (ClassNotFoundException e) {
                  log.error("Error while trying to perform search operation.", e);
                  throw new IndexingServiceException("Class type <" + className + "> not found.", e);
              }
          }
      }
  
      public String[] getTypesFromClass(Class<?> clazz) {
          List<String> types = Lists.newArrayList();
          Collection<Class<?>> allManagedClasses = typesToClasses.values();
          for (Class<?> managedClass : allManagedClasses) {
              if (clazz.isAssignableFrom(managedClass)) {
                  types.add(MappingBuilder.indexTypeFromClass(managedClass));
              }
          }
          return types.toArray(new String[types.size()]);
      }
  
      /**
       * Get the elastic search client linked to the index mapper.
       *
       * @return The elastic search client linked to the index mapper.
       */
      public Client getClient() {
          return this.esClient.getClient();
      }
  
      public static org.slf4j.Logger getLog() {
          return log;
      }
  
      private class TTL {
  
         protected String index;
         protected String field;
         protected String ttl;
  
         protected TTL (String index, String field, String ttl) {
            this.index = index;
            this.field = field;
            this.ttl = ttl;
         }
      } 
  
  }
  