D-Net

package eu.dnetlib.msro.workers.aggregation.store;

import java.util.logging.Level;

import java.util.logging.Logger;

import javax.annotation.PostConstruct;

import javax.annotation.PreDestroy;

import de.flapdoodle.embed.mongo.Command;

import de.flapdoodle.embed.mongo.MongodExecutable;

import de.flapdoodle.embed.mongo.MongodProcess;

import de.flapdoodle.embed.mongo.MongodStarter;

import de.flapdoodle.embed.mongo.config.*;

import de.flapdoodle.embed.mongo.distribution.Version;

import de.flapdoodle.embed.process.config.IRuntimeConfig;

import de.flapdoodle.embed.process.config.io.ProcessOutput;

import de.flapdoodle.embed.process.io.Processors;

import de.flapdoodle.embed.process.io.progress.LoggingProgressListener;

import de.flapdoodle.embed.process.runtime.Network;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.boot.autoconfigure.EnableAutoConfiguration;

import org.springframework.context.annotation.ComponentScan;

import org.springframework.context.annotation.Configuration;

import org.springframework.context.annotation.Profile;

import org.springframework.stereotype.Component;

/**

 * Created by claudio on 17/03/2017.

 */

@Profile("mdstoreTest")

@Component

@Configuration

@EnableAutoConfiguration

@ComponentScan(basePackages = "eu.dnetlib.msro.workers.aggregation.store")

public class EmbeddedMongodbConfiguration {

	private static final Log log = LogFactory.getLog(EmbeddedMongodbConfiguration.class);

	/**

	 * please store Starter or RuntimeConfig in a static final field

	 * if you want to use artifact store caching (or else disable caching)

	 */

	private MongodStarter starter;

	private MongodExecutable mongodExe;

	private MongodProcess mongod;

	private IMongodConfig mongodConfig;

	@Value("${msro.worker.mdstore.mongodb.port}")

	private int port;

	@PostConstruct

	public void init() throws Exception {

		log.info("starting mongodb embedded instance");

		Logger logger = Logger.getLogger(getClass().getName());

		ProcessOutput processOutput = new ProcessOutput(

				Processors.logTo(logger, Level.INFO),

				Processors.logTo(logger, Level.SEVERE),

				Processors.named("[console>]", Processors.logTo(logger, Level.FINE)));

		IRuntimeConfig runtimeConfig = new RuntimeConfigBuilder()

				.defaultsWithLogger(Command.MongoD,logger)

				.processOutput(processOutput)

				.artifactStore(new ExtractedArtifactStoreBuilder()

						.defaults(Command.MongoD)

						.download(new DownloadConfigBuilder()

								.defaultsForCommand(Command.MongoD)

								.progressListener(new LoggingProgressListener(logger, Level.FINE))))

				.build();

		mongodConfig = new MongodConfigBuilder()

				.version(Version.Main.PRODUCTION)

				.net(new Net(port, Network.localhostIsIPv6()))

				.build();

		starter = MongodStarter.getInstance(runtimeConfig);

		mongodExe = starter.prepare(mongodConfig);

		mongod = mongodExe.start();

	}

	@PreDestroy

	public void tearDown() throws Exception {

		log.info("stopping mongodb embedded instance");

		mongod.stop();

		mongodExe.stop();

	}

}

package eu.dnetlib.msro.workers.aggregation.store;

import org.junit.Assert;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.ActiveProfiles;

import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**

 * Created by claudio on 10/03/2017.

 */

@ActiveProfiles("mdstoreTest")

@RunWith(SpringJUnit4ClassRunner.class)

@SpringBootTest(classes = EmbeddedMongodbConfiguration.class)

public class MDStoreTest {

	@Autowired

	private MDStoreDao dao;

	@Test

	public void testMDStoreCreate() throws MDStoreException {

		Assert.assertNotNull(dao);

		final String mdId = "id";

		dao.createMDStore(mdId, "format", "layout", "interpretation");

		final MDStore mdstore = dao.getMDStore(mdId);

		Assert.assertNotNull(mdstore);

	}

}

msro.worker.mdstore.mongodb.port=27018

import io.prometheus.client.exporter.MetricsServlet;

import io.prometheus.client.hotspot.DefaultExports;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;

import org.springframework.boot.web.servlet.ServletRegistrationBean;

	@Bean

	@ConditionalOnMissingBean(name = "prometheusMetricsServletRegistrationBean")

	ServletRegistrationBean prometheusMetricsServletRegistrationBean(@Value("${prometheus.metrics.path:/prometheus}") String metricsPath) {

		DefaultExports.initialize();

		return new ServletRegistrationBean(new MetricsServlet(), metricsPath);

	}

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

package eu.dnetlib.msro.workers.aggregation.store;

import java.util.Date;

/**

 * Created by claudio on 13/03/2017.

 */

public class MDStoreTransactionInfo {

	private String id;

	private Boolean refresh;

	private Date date;

	private long size;

	/**

	 * @return the date

	 */

	public Date getDate() {

		return date;

	}

	/**

	 * @param date the date to set

	 */

	public void setDate(final Date date) {

		this.date = date;

	}

	/**

	 * @return the id

	 */

	public String getId() {

		return id;

	}

	/**

	 * @param id the id to set

	 */

	public void setId(final String id) {

		this.id = id;

	}

	/**

	 * @return the refresh

	 */

	public Boolean getRefresh() {

		return refresh;

	}

	/**

	 * @param refresh the refresh to set

	 */

	public void setRefresh(final Boolean refresh) {

		this.refresh = refresh;

	}

	/**

	 * @return the size

	 */

	public long getSize() {

		return size;

	}

	/**

	 * @param size the size to set

	 */

	public void setSize(final long size) {

		this.size = size;

	}

}

package eu.dnetlib.msro.workers.aggregation.store.mongo;

import java.time.Duration;

import java.time.LocalDate;

import java.time.ZoneId;

import java.util.Collections;

import java.util.Date;

import java.util.List;

import java.util.stream.Collectors;

import com.google.common.collect.Lists;

import com.mongodb.*;

import com.mongodb.client.FindIterable;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.MongoIterable;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.UpdateOptions;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

import eu.dnetlib.msro.workers.aggregation.store.*;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.conversions.Bson;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

/**

 * Created by claudio on 13/03/2017.

 */

@Component

public class MDStoreTransactionManagerImpl implements MDStoreTransactionManager {

	/**

	 * The Constant log.

	 */

	private static final Log log = LogFactory.getLog(MDStoreTransactionManagerImpl.class);

	/**

	 * The table name.

	 */

	private static String TABLE_NAME = "metadataManager";

	/**

	 * The max number of concurrent transactions per mdstore.

	 */

	private int maxTransactions = 1;

	/**

	 * The db.

	 */

	@Autowired

	private MongoDatabase db;

	/**

	 * The manager table.

	 */

	private MongoCollection<DBObject> managerTable;

	/**

	 * The expired days.

	 */

	private int expiredDays;

	/**

	 * Bootstrap manager.

	 */

	private void bootstrapManager() {

		log.debug("Bootstrap Manager start");

		final MongoCollection<DBObject> metadataColl = db.getCollection("metadata", DBObject.class);

		final FindIterable<DBObject> values = metadataColl.find();

		this.setManagerTable(db.getCollection(TABLE_NAME, DBObject.class));

		for (DBObject object : values) {

			final String id = (String) object.get("mdId");

			String newId = id;

			if (id.contains("_")) {

				newId = StringUtils.substringBefore(id, "_");

			}

			final DBObject input = BasicDBObjectBuilder.start()

				.add("mdId", id)

				.add("currentId", newId)

				.add("expiring", new String[] {})

				.add("transactions", new String[] {})

				.get();

			getManagerTable().insertOne(input);

			log.debug(String.format("Added %s to Metadata Manager data structure", id));

		}

		final BasicDBObject ensureIndex = new BasicDBObject();

		ensureIndex.put("mdId", 1);

		log.debug("Create index in MetadaManager ");

		this.getManagerTable().createIndex(ensureIndex);

	}

	/**

	 * Gets the DBObject describing an mdstore. null if there is no mdstore with the given id.

	 *

	 * @param mdstoreID

	 * @return DBObject or null

	 */

	private DBObject getMDStoreAsDBObject(String mdstoreID) {

		final BasicDBObject query = new BasicDBObject();

		query.put("mdId", mdstoreID);

		final FindIterable<DBObject> it = this.getManagerTable().find(query);

		DBObject mdstoreInfo = it.first();

		return mdstoreInfo;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void verifyConsistency() throws MDStoreException {

		if (this.getManagerTable() == null) {

			final int size = DnetStreamSupport.generateStreamFromIterator(

					db.listCollectionNames().iterator())

					.filter(it -> it.contains(TABLE_NAME))

					.collect(Collectors.toList())

					.size();

			if (size == 0)

				bootstrapManager();

			else {

				this.setManagerTable(db.getCollection(TABLE_NAME, DBObject.class));

			}

		}

		if (this.getManagerTable() == null) throw new MDStoreException("Something bad happen, unable to create managerTable");

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see MDStoreTransactionManager#createMDStore(String)

	 */

	@Override

	public void createMDStore(final String mdId) throws MDStoreException {

		log.debug("Creating new mdstore");

		verifyConsistency();

		String newId = mdId;

		if (mdId.contains("_")) {

			newId = StringUtils.substringBefore(mdId, "_");

		}

		final BasicDBObject instance = new BasicDBObject();

		instance.put("mdId", mdId);

		instance.put("currentId", newId);

		instance.put("expiring", new String[] {});

		getManagerTable().insertOne(instance);

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see MDStoreTransactionManager#dropMDStore(String)

	 */

	@Override

	public void dropMDStore(final String mdId) throws MDStoreException {

		verifyConsistency();

		log.debug("Droping MDStore: " + mdId);

		final BasicDBObject query = new BasicDBObject();

		query.put("mdId", mdId);

		final DBObject dropped = this.getManagerTable().findOneAndDelete(query);

		garbage();

		final String collectionName = (String) dropped.get("currentId");

		this.db.getCollection(collectionName).drop();

		this.db.getCollection("discarded-" + collectionName).drop();

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see MDStoreTransactionManager#getMDStoreCollection(String)

	 */

	@Override

	public String getMDStoreCollection(final String mdId) throws MDStoreException {

		verifyConsistency();

		DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo != null)

			return (String) mdstoreInfo.get("currentId");

		else return null;

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see MDStoreTransactionManager#startTransaction(String, boolean)

	 */

	@Override

	public String startTransaction(final String mdId, final boolean refresh) throws MDStoreException {

		verifyConsistency();

		log.info("Start transaction for metadata store " + mdId);

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo == null) throw new MDStoreException("Error, unable to find Mdstore with Id " + mdId);

		String idCreation = StringUtils.substringBefore(mdId, "_");

		idCreation = idCreation + "::" + System.currentTimeMillis();

		BasicDBList values;

		if (mdstoreInfo.containsField("transactions")) {

			values = (BasicDBList) mdstoreInfo.get("transactions");

			if (values.size() > getMaxTransactions())

				throw new MDStoreException(String.format("Cannot create more than %s transactions, found: %s, mdId: %s", getMaxTransactions(), values.size(), mdId));

		} else {

			values = new BasicDBList();

		}

		final BasicDBObject transactionMetadata = new BasicDBObject();

		transactionMetadata.put("id", idCreation.toString());

		transactionMetadata.put("refresh", refresh);

		transactionMetadata.put("date", new Date());

		values.add(transactionMetadata);

		mdstoreInfo.put("transactions", values);

		getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

		return idCreation.toString();

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see MDStoreTransactionManager#commit(String, String,

	 * MDStore)

	 */

	@Override

	public boolean commit(final String transactionId, final String mdstoreId, final MDStore current) throws MDStoreException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdstoreId);

		if (mdstoreInfo == null) throw new  MDStoreException("Error, unable to find Mdstore with Id " + mdstoreId);

		final BasicDBList transactions = (BasicDBList) mdstoreInfo.get("transactions");

		final DBObject transaction = findTransaction(transactions, transactionId);

		if (transaction == null) throw new  MDStoreException("Error, unable to find transaction with Id " + transactionId);

		final boolean refresh = (Boolean) transaction.get("refresh");

		transactions.remove(transaction);

		final String oldId = (String) mdstoreInfo.get("currentId");

		if (refresh) {

			mdstoreInfo.put("currentId", transactionId);

			final BasicDBList stillUsed = (BasicDBList) mdstoreInfo.get("expiring");

			if (stillUsed.size() == 0) {

				db.getCollection(oldId).drop();

				db.getCollection("discarded-" + oldId).drop();

			}

			log.debug("Replaced collection ");

		} else {

			log.debug("commit incremental ");

			updateIncremental(transactionId, oldId);

			db.getCollection(transactionId).drop();

			db.getCollection("discarded-" + transactionId).drop();

		}

		this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

		log.info("Committed transaction for metadata store " + mdstoreId);

		return true;

	}

	/**

	 * Find transaction.

	 *

	 * @param transactions  the transactions

	 * @param transactionId the transaction id

	 * @return the DB object

	 */

	private DBObject findTransaction(final BasicDBList transactions, final String transactionId) {

		if (transactions.size() == 0) return null;

		for (int i = 0; i < transactions.size(); i++) {

			final BasicDBObject transaction = (BasicDBObject) transactions.get(i);

			final String id = (String) transaction.get("id");

			if (transactionId.equals(id)) return transaction;

		}

		return null;

	}

	/**

	 * {@inheritDoc}

	 *

	 * @see MDStoreTransactionManager#readMdStore(String)

	 */

	@Override

	public String readMdStore(final String mdStoreId) throws  MDStoreException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdStoreId);

		if (mdstoreInfo == null) throw new  MDStoreException("Error, unable to find Mdstore with Id " + mdStoreId);

		final String currentId = (String) mdstoreInfo.get("currentId");

		final BasicDBList values = (BasicDBList) mdstoreInfo.get("expiring");

		updateMdstoreUsed(values, currentId);

		this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

		return currentId;

	}

	/**

	 * Update mdstore used.

	 *

	 * @param values the values

	 * @param mdId   the md id

	 */

	private void updateMdstoreUsed(final BasicDBList values, final String mdId) {

		if (values.size() > 0) {

			for (int i = 0; i < values.size(); i++) {

				final DBObject obj = (DBObject) values.get(i);

				final String id = (String) obj.get("id");

				if (mdId.equals(id)) {

					obj.put("lastRead", new Date());

					return;

				}

			}

		}

		final BasicDBObject readStore = new BasicDBObject();

		readStore.put("id", mdId);

		readStore.put("lastRead", new Date());

		values.add(readStore);

	}

	/**

	 * Gets the manager table.

	 *

	 * @return the managerTable

	 */

	public MongoCollection<DBObject> getManagerTable() {

		return managerTable;

	}

	/**

	 * Sets the manager table.

	 *

	 * @param managerTable the managerTable to set

	 */

	public void setManagerTable(final MongoCollection<DBObject> managerTable) {

		this.managerTable = managerTable;

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#getInfoForCurrentMdStore(java.lang.String)

	 */

	@Override

	public MDStoreManagerInfo getInfoForCurrentMdStore(final String mdStoreId) throws  MDStoreException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdStoreId);

		if (mdstoreInfo == null) throw new  MDStoreException("Error, unable to find Mdstore with Id " + mdStoreId);

		final MDStoreManagerInfo result = new MDStoreManagerInfo();

		result.setCurrentId((String) mdstoreInfo.get("currentId"));

		result.setMdId((String) mdstoreInfo.get("mdId"));

		final BasicDBList values = (BasicDBList) mdstoreInfo.get("expiring");

		for (int i = 0; i < values.size(); i++) {

			final MDStoreExpiredInfo stillused = new MDStoreExpiredInfo();

			final DBObject value = (DBObject) values.get(i);

			stillused.setId((String) value.get("id"));

			stillused.setLastRead((Date) value.get("lastRead"));

			result.addExpiredItem(stillused);

		}

		final BasicDBList transactions = (BasicDBList) mdstoreInfo.get("transactions");

		if (transactions != null) {

			for (int i = 0; i < transactions.size(); i++) {

				final MDStoreTransactionInfo transaction = new MDStoreTransactionInfo();

				final DBObject value = (DBObject) transactions.get(i);

				final String transactionId = (String) value.get("id");

				transaction.setId(transactionId);

				transaction.setDate((Date) value.get("date"));

				transaction.setRefresh((Boolean) value.get("refresh"));

				transaction.setSize(db.getCollection(transactionId).count());

				result.addTransactionInfo(transaction);

			}

		}

		return result;

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#dropUsed(java.lang.String, java.lang.String)

	 */

	@Override

	public Boolean dropUsed(final String mdId, final String idToDrop) throws  MDStoreException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo == null) throw new  MDStoreException("Error, unable to find Mdstore with Id " + mdId);

		return dropStore(mdstoreInfo, idToDrop, "expiring");

	}

	private boolean dropStore(DBObject mdstoreInfo, final String idToDrop, String transactionListName) throws  MDStoreException {

		final BasicDBList transactionList = (BasicDBList) mdstoreInfo.get(transactionListName);

		for (int i = 0; i < transactionList.size(); i++) {

			final DBObject value = (DBObject) transactionList.get(i);

			final String currentUsedId = (String) value.get("id");

			if (currentUsedId.equals(idToDrop)) {

				db.getCollection(idToDrop).drop();

				db.getCollection("discarded-" + idToDrop).drop();

				transactionList.remove(value);

				mdstoreInfo.put(transactionListName, transactionList);

				this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", mdstoreInfo.get("_id")), mdstoreInfo);

				return true;

			}

		}

		throw new  MDStoreException("Error, unable to drop collection " + idToDrop);

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#garbage()

	 */

	@Override

	public void garbage() throws  MDStoreException {

		verifyConsistency();

		log.info("Start garbage collection of MdStore");

		final FindIterable<DBObject> it = this.managerTable.find();

		int totalDeleted = 0;

		for (DBObject currentObject : it) {

			if (log.isDebugEnabled()) {

				log.debug("start to check id: " + currentObject.get("currentId"));

			}

			garbageExpiring(currentObject, (String) currentObject.get("currentId"));

			garbageTransactions(currentObject, (String) currentObject.get("currentId"));

			this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", currentObject.get("_id")), currentObject);

		}

		// DELETING Collection that are not in the metadataManager table

		MongoIterable<String> collections = this.db.listCollectionNames();

		for (String collection : collections) {

			if ((collection.length() > 30) && (collection.contains("discarded-") == false) && (collection.contains("resolved") == false)) {

				final DBObject item = getMetadataObjectForCollections(collection);

				if (shouldDelete(collection, item)) {

					if (log.isDebugEnabled()) {

						log.debug("delete collection: " + collection + " from mongo");

					}

					db.getCollection(collection).drop();

					db.getCollection("discarded-" + collection).drop();

					if (log.isDebugEnabled()) {

						log.debug("delete collection: discarded-" + collection + " from mongo");

					}

				}

			}

		}

		log.info("Complete garbage collection of MdStore, total store deleted: " + totalDeleted);

	}

	private DBObject getMetadataObjectForCollections(final String collectionName) {

		if (collectionName == null) return null;

		final String postfix = "_TURTdG9yZURTUmVzb3VyY2VzL01EU3RvcmVEU1Jlc291cmNlVHlwZQ==";

		final String tmp = collectionName.contains("discarded-") == true ? StringUtils.substringAfter(collectionName, "discarded-") : collectionName;

		final String collectionNameCleaned = StringUtils.substringBefore(tmp, "::") + postfix;

		//DBObject query = QueryBuilder.start("mdId").is(collectionNameCleaned).get();

		Bson query = Filters.eq("mdId", collectionNameCleaned);

		return this.managerTable.find(query).first();

	}

	private boolean shouldDelete(final String collectionName, final DBObject metadataManagerInstance) {

		log.debug("should delete instance " + metadataManagerInstance);

		if ((metadataManagerInstance == null) || (metadataManagerInstance.get("currentId") == null)) {

			log.debug("the instance has not currentID");

			return true;

		}

		String currentId = (String) metadataManagerInstance.get("currentId");

		if (collectionName.equals(currentId)) return false;

		BasicDBList expiringList = (BasicDBList) metadataManagerInstance.get("expiring");

		if (findInList(expiringList, collectionName, "id") == true) return false;

		BasicDBList transactionsList = (BasicDBList) metadataManagerInstance.get("transactions");

		return findInList(transactionsList, collectionName, "id") != true;

	}

	private boolean findInList(final BasicDBList list, final String object, final String tagname) {

		if (list == null) return false;

		for (int i = 0; i < list.size(); i++) {

			DBObject currentObject = (DBObject) list.get(i);

			final String id = (String) currentObject.get(tagname);

			if (id.equals(object)) return true;

		}

		return false;

	}

	/**

	 * Delete.

	 *

	 * @param list     the list

	 * @param toRemove the to remove

	 */

	private void delete(final BasicDBList list, final List<DBObject> toRemove) {

		for (final DBObject obj : toRemove) {

			if (log.isDebugEnabled()) {

				log.debug("deleting " + obj);

			}

			list.remove(obj);

		}

	}

	/**

	 * Garbage transactions.

	 *

	 * @param currentObject the current object

	 * @param currentId     the current id

	 */

	private void garbageTransactions(final DBObject currentObject, final String currentId) {

		if (log.isDebugEnabled()) {

			log.debug("Start garbage transactions ");

		}

		final BasicDBList expiring = (BasicDBList) currentObject.get("transactions");

		if ((expiring == null) || (expiring.size() <= getMaxTransactions())) return;

		List<DBObject> expiringList = Lists.newArrayList();

		for (int i = 0; i < expiring.size(); i++) {

			final DBObject cobj = (DBObject) expiring.get(i);

			if (cobj != null) {

				expiringList.add((DBObject) expiring.get(i));

			}

		}

		Collections.sort(expiringList, MDStoreUtils.getComparatorOnDate());

		List<DBObject> toRemove = Lists.newArrayList();

		int i = 0;

		// We should remove the k item less recent

		// where k = numberOftotalTransaction - maxNumberOfTransaction

		// k = numberOfItemToRemove

		while (((expiringList.size() - toRemove.size()) > getMaxTransactions()) || (i < expiringList.size())) {

			DBObject currentObj = expiringList.get(i++);

			String objectId = (String) currentObj.get("id");

			if (!objectId.equals(currentId)) {

				if (log.isDebugEnabled()) {

					log.debug("delete collection: " + objectId + " from mongo");

				}

				db.getCollection(objectId).drop();

				db.getCollection("discarded-" + objectId).drop();

				if (log.isDebugEnabled()) {

					log.debug("delete collection: discarded-" + objectId + " from mongo");

				}

				toRemove.add(currentObj);

			} else {

				if (log.isDebugEnabled()) {

					log.debug("Cannot remove transaction " + objectId + " because is the currentId: " + currentId);

				}

			}

		}

		delete(expiring, toRemove);

		log.info("Deleted " + toRemove.size() + " transactions, mdStore Id:" + currentObject.get("mdId"));

	}

	/**

	 * Garbage expiring.

	 *

	 * @param currentObject the current object

	 * @param currentId     the current id

	 */

	private void garbageExpiring(final DBObject currentObject, final String currentId) {

		if (log.isDebugEnabled()) {

			log.debug("Start to search expiring mdstores for id: " + currentObject.get("mdId"));

		}

		final BasicDBList expiring = (BasicDBList) currentObject.get("expiring");

		final List<DBObject> toRemove = Lists.newArrayList();

		if (log.isDebugEnabled()) {

			if (expiring == null) {

				log.debug("expiring list is null");

			} else {

				log.debug("expiring list size is :" + expiring.size());

			}

		}

		if ((expiring == null) || (expiring.size() == 0)) {

			log.debug("Deleted  0  expired  collections, mdStore Id:" + currentObject.get("mdId"));

			return;

		}

		for (int i = 0; i < expiring.size(); i++) {

			final DBObject currentExpiredStore = (DBObject) expiring.get(i);

			final String currentUsedId = (String) currentExpiredStore.get("id");

			final long d = getExpiringDays(currentExpiredStore, "lastRead");

			if (log.isDebugEnabled()) {

				log.debug("the store :" + currentId + " expired since " + d + "days ");

			}

			// DELETE the collection where the last time they was read

			// is more than 3 days ago

			if (d > getExpiredDays()) {

				if (currentUsedId.equals(currentId) == false) {

					db.getCollection(currentUsedId).drop();

					db.getCollection("discarded-" + currentUsedId).drop();

					log.debug("deleted collection " + currentUsedId);

				}

				toRemove.add(currentExpiredStore);

			}

		}

		delete(expiring, toRemove);

		log.debug("Deleted expired " + toRemove.size() + "collections, mdStore Id:" + currentObject.get("mdId"));

	}

	/**

	 * Gets the expiring days.

	 *

	 * @param value     the value

	 * @param paramName the param name

	 * @return the expiring days

	 */

	private long getExpiringDays(final DBObject value, final String paramName) {

		final Date lastRead = (Date) value.get(paramName);

		final LocalDate readDate = lastRead.toInstant().atZone(ZoneId.systemDefault()).toLocalDate();

		return Duration.between(LocalDate.now().atTime(0, 0), readDate.atTime(0, 0)).toDays();

	}

	/**

	 * Gets the expired days.

	 *

	 * @return the expiredDays

	 */

	public int getExpiredDays() {

		if (this.expiredDays == 0) return 3;

		return expiredDays;

	}

	/**

	 * Sets the expired days.

	 *

	 * @param expiredDays the expiredDays to set

	 */

	public void setExpiredDays(final int expiredDays) {

		this.expiredDays = expiredDays;

	}

	/**

	 * Update incremental.

	 *

	 * @param transactionId the transaction id

	 * @param currentId     the current id

	 */

	private void updateIncremental(final String transactionId, final String currentId) {

		final MongoCollection<DBObject> transaction = db.getCollection(transactionId, DBObject.class);

		final MongoCollection<DBObject> mdstore = db.getCollection(currentId, DBObject.class);

		final FindIterable<DBObject> it = transaction.find();

		for (DBObject currentObj : it) {

			BasicDBObject newObj = new BasicDBObject();

			final String id = (String) currentObj.get("id");

			final String body = (String) currentObj.get("body");

			newObj.put("id", id);

			newObj.put("body", body);

			if (StringUtils.isNotBlank(id)) {

				//setting to journaled write concern to be sure that when the write returns everything has been flushed to disk (https://docs.mongodb.org/manual/faq/developers/#when-does-mongodb-write-updates-to-disk)

				//the explicit fsync command can't be run anymore: 'Command failed with error 13: 'fsync may only be run against the admin database.'

				final MongoCollection<DBObject> mdstoreWrite = mdstore.withWriteConcern(WriteConcern.JOURNALED);

				mdstoreWrite.replaceOne(new BasicDBObject("id", id), newObj, new UpdateOptions().upsert(true));

			}

		}

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.data.mdstore.modular.connector.MDStoreTransactionManager#dropTransaction(java.lang.String, java.lang.String)

	 */

	@Override

	public Boolean dropTransaction(final String mdId, final String idToDrop) throws  MDStoreException {

		verifyConsistency();

		final DBObject mdstoreInfo = getMDStoreAsDBObject(mdId);

		if (mdstoreInfo == null) throw new  MDStoreException("Error, unable to find Mdstore with Id " + mdId);

		return dropStore(mdstoreInfo, idToDrop, "transactions");

	}

	public void garbageTransactionsOnStart() throws  MDStoreException {

		verifyConsistency();

		FindIterable<DBObject> it = this.managerTable.find();

		for (DBObject currentObject : it) {

			final BasicDBList transactions = (BasicDBList) currentObject.get("transactions");

			if ((transactions != null) && (transactions.size() > 0)) {

				for (int i = 0; i < transactions.size(); i++) {

					final DBObject currentTransactions = (DBObject) transactions.get(i);

					final String id = (String) currentTransactions.get("id");

					db.getCollection(id).drop();

					db.getCollection("discarded-" + id).drop();

					log.debug("deleted collection " + id);

				}

				currentObject.put("transactions", new BasicDBList());

				this.getManagerTable().findOneAndReplace(new BasicDBObject("_id", currentObject.get("_id")), currentObject);

			}

		}

	}

	/**

	 * Gets the max transactions.

	 *

	 * @return the maxTransactions

	 */

	public int getMaxTransactions() {

		return maxTransactions;

	}

	/**

	 * Sets the max transactions.

	 *

	 * @param maxTransactions the maxTransactions to set

	 */

	public void setMaxTransactions(final int maxTransactions) {

		this.maxTransactions = maxTransactions;

	}

}

package eu.dnetlib.msro.workers.aggregation.store.mongo;

import org.springframework.boot.context.properties.ConfigurationProperties;

import org.springframework.stereotype.Component;

/**

 * Created by claudio on 13/03/2017.

 */

@Component

@ConfigurationProperties(prefix = "msro.worker.mdstore.mongodb")

public class MongoDBProperties {

	private String host = "localhost";

	private int port = 27017;

	private String dbname = "mdstore";

	public String getHost() {

		return host;

	}

	public void setHost(final String host) {

		this.host = host;

	}

	public int getPort() {

		return port;

	}

	public void setPort(final int port) {

		this.port = port;

	}

	public String getDbname() {

		return dbname;

	}

	public void setDbname(final String dbname) {

		this.dbname = dbname;

	}

}

package eu.dnetlib.msro.workers.aggregation.store.mongo;

import java.util.List;

import java.util.stream.Collectors;

import com.google.common.collect.Lists;

import com.google.common.primitives.Ints;

import com.mongodb.BasicDBObject;

import com.mongodb.BasicDBObjectBuilder;

import com.mongodb.DBObject;

import com.mongodb.client.AggregateIterable;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.CreateCollectionOptions;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

import eu.dnetlib.msro.workers.aggregation.store.*;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

/**

 * Created by claudio on 13/03/2017.

 */

@Component

public class MDStoreDaoImpl implements MDStoreDao {

	private static final Log log = LogFactory.getLog(MDStoreDaoImpl.class);

	public static final String MD_ID = "mdId";

	public static final String FORMAT = "format";

	public static final String INTERPRETATION = "interpretation";

	public static final String LAYOUT = "layout";

	public static final String SIZE = "size";

	public static final String METADATA_NAME = "metadata";

	private boolean discardRecords = true;

	@Autowired

	private MongoDatabase db;

	@Autowired

	private MDStoreTransactionManager transactionManager;

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void createMDStore(final String mdId, final String format, final String interpretation, final String layout) throws MDStoreException {

		transactionManager.createMDStore(mdId);

		final String internalId = transactionManager.getMDStoreCollection(mdId);

		final List<String> names = DnetStreamSupport.generateStreamFromIterator(db.listCollectionNames().iterator())

				.filter(c -> c.contains(internalId))

				.collect(Collectors.toList());

		if (names.isEmpty()) {

			log.info(String.format("creating collection %s", internalId));

			db.createCollection(internalId, new CreateCollectionOptions());

		}

		final BasicDBObject obj = new BasicDBObject();

		obj.put(MD_ID, mdId);

		obj.put(FORMAT, format);

		obj.put(INTERPRETATION, interpretation);

		obj.put(LAYOUT, layout);

		obj.put(SIZE, 0);

		getMetadataCollection().insertOne(obj);

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void deleteMDStore(final String mdId) throws MDStoreException {

		final MongoCollection<DBObject> metadata = getMetadataCollection();

		if (metadata == null) throw new MDStoreException("cannot find metadata collection");

		transactionManager.dropMDStore(mdId);

		metadata.deleteOne(new BasicDBObject(MD_ID, mdId));

		log.info("deleted mdId: " + mdId);

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public MDStore getMDStore(final String mdId) throws MDStoreException {

		final String internalId = transactionManager.getMDStoreCollection(mdId);

		return new MongoMDStore(mdId, getRecordCollection(internalId), isDiscardRecords(), db);

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public MDStore readMDStore(final String mdId) throws MDStoreException {

		final String internalId = transactionManager.readMdStore(mdId);

		return new MongoMDStore(mdId, getRecordCollection(internalId), isDiscardRecords(), db);

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public MDStore startTransaction(final String mdId, final boolean refresh) throws MDStoreException {

		final String transactionId = transactionManager.startTransaction(mdId, refresh);

		return new MongoMDStore(transactionId, getRecordCollection(transactionId), isDiscardRecords(), db);

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public List<MDStoreDescription> listMDStores() {

		return DnetStreamSupport.generateStreamFromIterator(getMetadataCollection().find().iterator())

				.map(this::convertMDStoreDescription)

				.collect(Collectors.toList());

	}

	private MongoCollection<Record> getRecordCollection(final String collectionId) {

		return db.getCollection(collectionId, Record.class);

	}

	private MongoCollection<DBObject> getMetadataCollection() {

		return db.getCollection(METADATA_NAME, DBObject.class);

	}

	private MDStoreDescription convertMDStoreDescription(DBObject input) {

		final String mdId = (String) input.get(MD_ID);

		log.debug("Getting info for " + mdId);

		final String format = (String) input.get(FORMAT);

		final String layout = (String) input.get(LAYOUT);

		final String interpretation = (String) input.get(INTERPRETATION);

		MongoMDStore currentMDStore = null;

		final MDStoreDescription description = new MDStoreDescription();

		try {

			currentMDStore = (MongoMDStore) getMDStore(mdId);

		} catch (final MDStoreException e) {

			log.error("Error on retrieving mdstore for getting info mdId " + mdId);

		}

		int size = 0;

		if (input.containsField(SIZE)) {

			log.debug("Size retrieved from metadata for mdId :" + mdId);

			size = (Integer) input.get(SIZE);

		} else {

			if (currentMDStore != null) {

				log.debug("Size not Found in metadata for mdId :" + mdId + " calling getCount ");

				size = currentMDStore.getSize();

				input.put("size", size);

				db.getCollection(METADATA_NAME, DBObject.class).findOneAndReplace(new BasicDBObject(MD_ID, mdId), input);

			}

		}

		if (currentMDStore != null) {

			description.setIndexed(currentMDStore.isIndexed());

		}

		description.setId(mdId);

		description.setFormat(format);

		description.setLayout(layout);

		description.setInterpretation(interpretation);

		description.setSize(size);

		return description;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public List<String> listMDStores(final String format, final String layout, final String interpretation) {

		return DnetStreamSupport.generateStreamFromIterator(getMetadataCollection().find().iterator())

				.filter(MDStoreUtils.dboFilter(format, layout, interpretation))

				.map(MDStoreUtils.mdId())

				.collect(Collectors.toList());

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public int getCachedSize(final String id) throws MDStoreException {

		log.debug("retrieve cached size for mdstore: " + id);

		final DBObject desc = getMetadataCollection().find(new BasicDBObject(MD_ID, id)).first();

		if (!desc.containsField(SIZE)) {

			desc.put(SIZE, getMDStore(id).getSize());

		}

		final Object oSize = desc.get(SIZE);

		return (Integer) oSize;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void refreshSizes() throws MDStoreException {

		for (final MDStoreDescription mdStoreId : listMDStores()) {

			refreshSize(mdStoreId.getId());

		}

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public int refreshSize(final String mdStoreId) throws MDStoreException {

		final int size = (int) db.getCollection(transactionManager.getMDStoreCollection(mdStoreId)).count();

		final MongoCollection<DBObject> metadata = db.getCollection(METADATA_NAME, DBObject.class);

		metadata.updateOne(new BasicDBObject(MD_ID, mdStoreId), new BasicDBObject("$set", new BasicDBObject(SIZE, size)));

		return size;

	}

	@Override

	public int getSumOfSizes(final String format, final String layout, final String interpretation) throws MDStoreException {

		final MongoCollection<DBObject> metadata = getMetadataCollection();

		BasicDBObject matchObj = (BasicDBObject) BasicDBObjectBuilder.start("$match",

				BasicDBObjectBuilder.start("format", format).add("layout", layout).add("interpretation", interpretation).get()).get();

		BasicDBObject groupObj = (BasicDBObject) BasicDBObjectBuilder.start("$group",

				BasicDBObjectBuilder.start("_id", "").add("total", new BasicDBObject("$sum", "$" + SIZE)).get()).get();

		BasicDBObject projectObj = new BasicDBObject("$project", new BasicDBObject("_id", 0).append("total", 1));

		List<BasicDBObject> pipeline = Lists.newArrayList(matchObj, groupObj, projectObj);

		AggregateIterable<DBObject> output = metadata.aggregate(pipeline, DBObject.class);

		DBObject result = output.first();

		if (result == null || !result.containsField("total")) {

			log.debug("No total found");

			return 0;

		} else return (Integer) result.get("total");

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public void commit(final String transactionId, final String mdId) throws MDStoreException {

		transactionManager.commit(transactionId, mdId, getMDStore(mdId));

	}

	/**

	 * Getter for property 'discardRecords'.

	 *

	 * @return Value for property 'discardRecords'.

	 */

	public boolean isDiscardRecords() {

		return discardRecords;

	}

	/**

	 * Setter for property 'discardRecords'.

	 *

	 * @param discardRecords Value to set for property 'discardRecords'.

	 */

	public void setDiscardRecords(final boolean discardRecords) {

		this.discardRecords = discardRecords;

	}

	@Override

	public MDStoreDBStatus getDBStatus() {

		final int handledDatastructures = Ints.saturatedCast(db.getCollection(METADATA_NAME).count());

		//final int usedDiskSpace = Ints.saturatedCast(db.getStats().getLong("storageSize") / (1024 * 1024)); // in MB

		//{ dbStats: 1, scale: 1 }

		BasicDBObject statsQuery = new BasicDBObject("dbStats", 1);

		statsQuery.put("scale", 1024 * 1024); //storageSize in MB

		final Document statsRes = db.runCommand(statsQuery);

		log.debug("DBStatus --  " + statsRes.toJson());

		int usedDiskSpace = 0;

		//trying to handle different versions of the mongo server: old version returns storage size as long, new version as double

		//TODO: simplify this when dev, beta, production are aligned with our local, latest, mongo version

		String usedDiskSpaceStr = statsRes.get("storageSize").toString();

		try {

			Long usedDiskSpaceLong = Long.parseLong(usedDiskSpaceStr);

			usedDiskSpace = Ints.saturatedCast(usedDiskSpaceLong);

		} catch (NumberFormatException nfe) {

			Double usedDiskSpaceDbl = Double.parseDouble(usedDiskSpaceStr);

			usedDiskSpace = usedDiskSpaceDbl.intValue();

		}

		final String date = DateUtils.now_ISO8601();

		return new MDStoreDBStatus(handledDatastructures, usedDiskSpace, date);

	}

	@Override

	public void startGarbage() throws MDStoreException {

		this.transactionManager.garbage();

	}

	@Override

	public void invalidTransaction(final String transactionId, final String mdId) throws MDStoreException {

		transactionManager.dropTransaction(mdId, transactionId);

	}

}

package eu.dnetlib.msro.workers.aggregation.store.mongo;

import java.util.List;

import com.google.common.collect.Lists;

import com.mongodb.BasicDBObject;

import com.mongodb.WriteConcern;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.model.*;

import eu.dnetlib.msro.workers.aggregation.store.Record;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class MongoBulkWritesManager {

	private static final Log log = LogFactory.getLog(MongoBulkWritesManager.class);

	private final boolean discardRecords;

	private MongoCollection<Record> validCollection;

	private List<WriteModel<Record>> validBulkOperationList;

	private BulkWriteOptions writeOptions;

	private MongoCollection<Record> discardedCollection;

	private List<WriteModel<Record>> discardedBulkOperationList;

	private int bulkSize;

	public MongoBulkWritesManager(final MongoCollection<Record> collection,

			final MongoCollection<Record> discardedCollection,

			final int bulkSize,

			final boolean discardRecords) {

		this.validCollection = collection.withWriteConcern(WriteConcern.ACKNOWLEDGED);

		this.validBulkOperationList = Lists.newArrayList();

		this.discardedCollection = discardedCollection.withWriteConcern(WriteConcern.ACKNOWLEDGED);

		this.discardedBulkOperationList = Lists.newArrayList();

		this.bulkSize = bulkSize;

		this.discardRecords = discardRecords;

		this.writeOptions = new BulkWriteOptions().ordered(false);

	}

	public void insert(final Record r) {

		try {

			if (StringUtils.isNotBlank(r.getId())) {

				if (log.isDebugEnabled()) {

					log.debug("Saving object" + r.getId());

				}

				validBulkOperationList.add(new ReplaceOneModel(new BasicDBObject("id", r.getId()), r, new UpdateOptions().upsert(true)));

				if (((validBulkOperationList.size() % bulkSize) == 0) && !validBulkOperationList.isEmpty()) {

					validCollection.bulkWrite(validBulkOperationList, writeOptions);

					validBulkOperationList.clear();

				}

			} else {

				if (discardRecords) {

					log.debug("parsed record seems invalid");

					discardRecord(r);

				}

			}

		} catch (Throwable e) {

			if (discardRecords) {

				log.debug("unhandled exception: " + e.getMessage());

				discardRecord(r);

			}

		}

	}

	private void discardRecord(final Record record) {

		discardedBulkOperationList.add(new InsertOneModel(new BasicDBObject("body", record)));

		if (((discardedBulkOperationList.size() % bulkSize) == 0) && !discardedBulkOperationList.isEmpty()) {

			discardedCollection.bulkWrite(discardedBulkOperationList, writeOptions);

			discardedBulkOperationList.clear();

		}

	}

	public void flushBulks() {

		//setting to journaled write concern to be sure that when the write returns everything has been flushed to disk (https://docs.mongodb.org/manual/faq/developers/#when-does-mongodb-write-updates-to-disk)

		//the explicit fsync command can't be run anymore: 'Command failed with error 13: 'fsync may only be run against the admin database.'

		if (!validBulkOperationList.isEmpty()) {

			validCollection = getCollectionWithWriteConcern(validCollection, WriteConcern.JOURNALED);

			validCollection.bulkWrite(validBulkOperationList, writeOptions);

		}

		if (!discardedBulkOperationList.isEmpty()) {

			discardedCollection = getCollectionWithWriteConcern(discardedCollection, WriteConcern.ACKNOWLEDGED);

			discardedCollection.bulkWrite(discardedBulkOperationList, writeOptions);

		}

		//setting write concern back to ACKNOWLEDGE to avoid the execution of future writes all in Journaled mode

		validCollection = getCollectionWithWriteConcern(validCollection, WriteConcern.ACKNOWLEDGED);

		discardedCollection = getCollectionWithWriteConcern(discardedCollection, WriteConcern.ACKNOWLEDGED);

	}

	private MongoCollection<Record> getCollectionWithWriteConcern(MongoCollection<Record> collection, WriteConcern writeConcern) {

		return collection.withWriteConcern(writeConcern);

	}

}

package eu.dnetlib.msro.workers.aggregation.store.mongo;

import com.mongodb.MongoClient;

import com.mongodb.client.MongoDatabase;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/**

 * Created by claudio on 13/03/2017.

 */

@Configuration

public class MongoDBConfiguration {

	private static final Log log = LogFactory.getLog(MongoDBConfiguration.class);

	@Autowired

	private MongoDBProperties properties;

	@Bean

	public MongoClient mongoClient() throws Exception {

		log.info("creating MongoClient");

		return new MongoClient(properties.getHost(), properties.getPort());

	}

	@Bean

	public MongoDatabase db() throws Exception {

		log.info("creating MongoDatabase");

		return mongoClient().getDatabase(properties.getDbname());

	}