.put("discovery.zen.ping.multicast.enabled", "false")
// Index storage policies
- .put("index.refresh_interval", "30")
.put("index.number_of_shards", "1")
.put("index.number_of_replicas", MINIMUM_INDEX_REPLICATION)
.put("index.store.type", "mmapfs")
-// .put("indices.store.throttle.type", "merge")
-// .put("indices.store.throttle.max_bytes_per_sec", "500mb")
- .put("index.merge.scheduler.max_thread_count",
- Math.max(1, Math.min(3, Runtime.getRuntime().availableProcessors() / 2)))
// Install our own listUpdate scripts
.put("script.default_lang", "native")
import org.sonar.server.search.IndexDefinition;
import org.sonar.server.search.action.DeleteKey;
import org.sonar.server.search.action.DeleteNestedItem;
-import org.sonar.server.search.action.RefreshActionRequest;
import org.sonar.server.search.action.UpsertDto;
import org.sonar.server.search.action.UpsertNestedItem;
import javax.annotation.CheckForNull;
import javax.annotation.Nullable;
+
import java.io.Serializable;
import java.sql.Timestamp;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import static com.google.common.collect.Maps.newHashMap;
@Override
public void handleResult(ResultContext resultContext) {
DTO dto = (DTO) resultContext.getResultObject();
- session.enqueue(new UpsertDto<DTO>(getIndexType(), dto, false));
+ session.enqueue(new UpsertDto<DTO>(getIndexType(), dto, true));
count++;
if (count % 100000 == 0) {
LOGGER.info(" - synchronized {} {}", count, getIndexType());
DbSynchronizationHandler handler = getSynchronizationResultHandler(session);
session.select(getSynchronizeStatementFQN(), getSynchronizationParams(date, params), handler);
handler.enqueueCollected();
- session.enqueue(new RefreshActionRequest(this.getIndexType()));
} catch (Exception e) {
throw new IllegalStateException(e);
}
return mapping;
}
+ @Override
+ public void refresh() {
+ client.execute(client
+ .admin()
+ .indices()
+ .prepareRefresh(this.getIndexName())
+ .setForce(false)
+ .setIndices(this.getIndexName()));
+ }
+
/* Base CRUD methods */
protected abstract DOMAIN toDoc(Map<String, Object> fields);
String getIndexName();
+ void refresh();
+
Date getLastSynchronization();
IndexStat getIndexStat();
package org.sonar.server.search;
import org.elasticsearch.action.ActionRequest;
-import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequestBuilder;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.sonar.api.ServerComponent;
import org.sonar.api.config.Settings;
import org.sonar.api.platform.ComponentContainer;
-import org.sonar.core.cluster.ClusterAction;
import org.sonar.core.cluster.WorkQueue;
import org.sonar.core.profiling.Profiling;
import org.sonar.server.search.action.IndexActionRequest;
-import org.sonar.server.search.action.IndexWorker;
-import org.sonar.server.search.action.RefreshActionRequest;
import java.util.HashMap;
import java.util.HashSet;
}
try {
- boolean refreshRequired = false;
-
Map<String, Index> indexes = getIndexMap();
Set<String> indices = new HashSet<String>();
- for (ClusterAction action : actions) {
- if (IndexActionRequest.class.isAssignableFrom(action.getClass())) {
- IndexActionRequest worker = (IndexActionRequest) action;
- if (worker.needsRefresh()) {
- refreshRequired = true;
- indices.add(indexes.get(worker.getIndexType()).getIndexName());
- }
- }
-
- if (IndexWorker.class.isAssignableFrom(action.getClass())) {
- IndexWorker worker = (IndexWorker) action;
- Index index = indexes.get(worker.getIndexType());
- worker.setIndex(index);
- }
-
- if (RefreshActionRequest.class.isAssignableFrom(action.getClass())) {
- refreshRequired = true;
- indices.add(indexes.get(((RefreshActionRequest) action).getIndexType()).getIndexName());
+ for (IndexActionRequest action : actions) {
+ Index index = indexes.get(action.getIndexType());
+ action.setIndex(index);
+ if (action.needsRefresh()) {
+ indices.add(index.getIndexName());
}
}
long normTime = executeNormalization(bulkRequestBuilder, actions);
- if (bulkRequestBuilder.numberOfActions() > 0) {
- // execute the request
- long indexTime = System.currentTimeMillis();
- BulkResponse response = searchClient.execute(bulkRequestBuilder.setRefresh(false));
+ //execute the request
+ long indexTime = System.currentTimeMillis();
+ BulkResponse response = searchClient.execute(bulkRequestBuilder.setRefresh(false));
+ indexTime = System.currentTimeMillis() - indexTime;
- indexTime = System.currentTimeMillis() - indexTime;
+ long refreshTime = this.refreshRequiredIndex(indices);
- long refreshTime = 0;
- if (refreshRequired) {
- refreshTime = this.refreshRequiredIndex(indices);
- }
-
- LOGGER.debug("-- submitted {} items with {}ms in normalization, {}ms indexing and {}ms refresh({}). Total: {}ms",
- bulkRequestBuilder.numberOfActions(), normTime, indexTime, refreshTime, indices, (normTime + indexTime + refreshTime));
+ LOGGER.debug("-- submitted {} items with {}ms in normalization, {}ms indexing and {}ms refresh({}). Total: {}ms",
+ bulkRequestBuilder.numberOfActions(), normTime, indexTime, refreshTime, indices, (normTime + indexTime + refreshTime));
- if (response.hasFailures()) {
- throw new IllegalStateException("Errors while indexing stack: " + response.buildFailureMessage());
- }
+ if (response.hasFailures()) {
+ throw new IllegalStateException("Errors while indexing stack: " + response.buildFailureMessage());
}
+
} catch (Exception e) {
LOGGER.error("Could not commit to ElasticSearch", e);
}
}
+
private long refreshRequiredIndex(Set<String> indices) {
long refreshTime = System.currentTimeMillis();
private long executeNormalization(BulkRequestBuilder bulkRequestBuilder, List<IndexActionRequest> actions) {
long normTime = System.currentTimeMillis();
try {
- boolean hasInlineRefreshRequest = false;
ExecutorService executorService = Executors.newFixedThreadPool(CONCURRENT_NORMALIZATION_FACTOR);
- // invokeAll() blocks until ALL tasks submitted to executor complete
+ //invokeAll() blocks until ALL tasks submitted to executor complete
for (Future<List<ActionRequest>> updateRequests : executorService.invokeAll(actions)) {
for (ActionRequest update : updateRequests.get()) {
if (UpdateRequest.class.isAssignableFrom(update.getClass())) {
- bulkRequestBuilder.add(((UpdateRequest) update));
+ bulkRequestBuilder.add(((UpdateRequest) update).refresh(false));
} else if (DeleteRequest.class.isAssignableFrom(update.getClass())) {
- bulkRequestBuilder.add(((DeleteRequest) update));
- } else if (RefreshRequest.class.isAssignableFrom(update.getClass())) {
- hasInlineRefreshRequest = true;
+ bulkRequestBuilder.add(((DeleteRequest) update).refresh(false));
} else {
throw new IllegalStateException("Un-managed request type: " + update.getClass());
}
}
}
executorService.shutdown();
- bulkRequestBuilder.setRefresh(hasInlineRefreshRequest);
} catch (Exception e) {
throw new IllegalStateException("Could not execute normalization for stack", e);
}
import org.sonar.core.profiling.StopWatch;
import org.sonar.process.LoopbackAddress;
-import java.io.File;
-
/**
* ElasticSearch Node used to connect to index.
*/
public void stop() {
close();
}
-
- public File executePayload(File f) {
-
- return new File("payload");
- }
}
*/
package org.sonar.server.search.action;
+
import org.elasticsearch.action.ActionRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.sonar.core.cluster.ClusterAction;
import java.util.ArrayList;
import java.util.List;
-public abstract class IndexActionRequest implements IndexWorker, ClusterAction<ActionRequest> {
+public abstract class IndexActionRequest implements ClusterAction<List<ActionRequest>> {
protected final String indexType;
private final boolean requiresRefresh;
}
protected IndexActionRequest(String indexType, boolean requiresRefresh) {
+ super();
this.indexType = indexType;
this.requiresRefresh = requiresRefresh;
}
return indexType;
}
+
public void setIndex(Index index) {
this.index = index;
}
((UpdateRequest) request)
.type(index.getIndexType())
.index(index.getIndexName())
- .refresh(requiresRefresh);
+ .refresh(false);
}
finalRequests.add(request);
}
+++ /dev/null
-/*
- * SonarQube, open source software quality management tool.
- * Copyright (C) 2008-2014 SonarSource
- * mailto:contact AT sonarsource DOT com
- *
- * SonarQube is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 3 of the License, or (at your option) any later version.
- *
- * SonarQube is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public License
- * along with this program; if not, write to the Free Software Foundation,
- * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
- */
-package org.sonar.server.search.action;
-
-import org.sonar.server.search.Index;
-
-public interface IndexWorker {
-
- public String getIndexType();
-
- public void setIndex(Index index);
-}
+++ /dev/null
-/*
- * SonarQube, open source software quality management tool.
- * Copyright (C) 2008-2014 SonarSource
- * mailto:contact AT sonarsource DOT com
- *
- * SonarQube is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Lesser General Public
- * License as published by the Free Software Foundation; either
- * version 3 of the License, or (at your option) any later version.
- *
- * SonarQube is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public License
- * along with this program; if not, write to the Free Software Foundation,
- * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
- */
-package org.sonar.server.search.action;
-
-import com.google.common.collect.ImmutableList;
-import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
-import org.sonar.core.cluster.ClusterAction;
-import org.sonar.server.search.Index;
-
-import java.util.List;
-
-public class RefreshActionRequest implements IndexWorker, ClusterAction<RefreshRequest> {
-
- private Index<?, ?, ?> index;
- private final String indexType;
-
- public RefreshActionRequest(String indexType) {
- this.indexType = indexType;
- }
-
- @Override
- public List<RefreshRequest> call() throws Exception {
- return ImmutableList.<RefreshRequest>of(
- new RefreshRequest()
- .force(false)
- .indices(index.getIndexName()));
- }
-
- @Override
- public String getIndexType() {
- return indexType;
- }
-
- @Override
- public void setIndex(Index index) {
- this.index = index;
- }
-}
@Test
public void massive_insert() {
- // Set qeue's implicit commit size to 10
- dbSession.setImplicitCommitSize(20);
-
// 0 Assert no logs in DB
assertThat(dao.findAll(dbSession)).hasSize(0);
- int max = 35;
+ int max = 400;
final String testValue = "hello world";
for (int i = 0; i < max; i++) {
@Test
public void massive_log_insert() {
- // Set qeue's implicit commit size to 10
- dbSession.setImplicitCommitSize(10);
-
// 0 Assert no logs in DB
assertThat(dao.findAll(dbSession)).hasSize(0);
- int max = 40;
+ int max = 400;
final String testValue = "hello world";
for (int i = 0; i < max; i++) {
TestActivityLog log = new TestActivityLog(testValue + "_" + i, Activity.Type.QPROFILE.toString());
assertThat(session.getActionCount()).isEqualTo(1);
dao.synchronizeAfter(session, new Date(t0));
- // Synchronize adds an implicit action to the queue before finishing.
- assertThat(session.getActionCount()).isEqualTo(3);
+ assertThat(session.getActionCount()).isEqualTo(2);
}
}
assertThat(session.getActionCount()).isEqualTo(0);
dao.synchronizeAfter(session, new Date(0));
- // SynchronizeAfter adds an implicit action (refresh) after execution of synchronization
- assertThat(session.getActionCount()).isEqualTo(2);
+
+ assertThat(session.getActionCount()).isEqualTo(1);
}
@Test
setupData("synchronize_after_since_given_date");
dao.synchronizeAfter(session, DateUtils.parseDate("2014-09-01"));
- // SynchronizeAfter adds an implicit action (refresh) after execution of synchronization
- assertThat(session.getActionCount()).isEqualTo(2);
+ assertThat(session.getActionCount()).isEqualTo(1);
}
@Test
setupData("synchronize_after_with_project");
dao.synchronizeAfter(session, DateUtils.parseDate("2014-01-01"), ImmutableMap.of(IssueAuthorizationDao.PROJECT_KEY, "org.sonar:sample"));
- // SynchronizeAfter adds an implicit action (refresh) after execution of synchronization
- assertThat(session.getActionCount()).isEqualTo(2);
+ assertThat(session.getActionCount()).isEqualTo(1);
}
}
synchronizer.synchronize(dbSession, dbClient.ruleDao(), indexClient.get(RuleIndex.class));
dbSession.commit();
- Thread.sleep(1000);
assertThat(indexClient.get(RuleIndex.class).countAll()).isEqualTo(numberOfRules);
}
}
.getState().getMetaData().concreteAllIndices())
.setQuery(QueryBuilders.matchAllQuery())
.get();
- searchClient.admin().indices().prepareRefresh(searchClient.admin().cluster().prepareState().get()
- .getState().getMetaData().concreteAllIndices())
- .setForce(true)
- .get();
- searchClient.admin().indices().prepareFlush(searchClient.admin().cluster().prepareState().get()
- .getState().getMetaData().concreteAllIndices())
- .get();
+
}
}
public void clearIndexes() {
checkStarted();
get(BackendCleanup.class).clearIndexes();
-
}
/**
*/
package org.sonar.core.cluster;
-import java.util.List;
import java.util.concurrent.Callable;
-public interface ClusterAction<K> extends Callable<List<K>> {
+public interface ClusterAction<K> extends Callable<K> {
@Override
- public List<K> call() throws Exception;
+ public K call() throws Exception;
}
private SqlSession session;
private int actionCount;
- private Integer implicitCommitSize = IMPLICIT_COMMIT_SIZE;
-
DbSession(WorkQueue queue, SqlSession session) {
this.actionCount = 0;
this.session = session;
this.actions = new ArrayList<ClusterAction>();
}
- public Integer getImplicitCommitSize() {
- return implicitCommitSize;
- }
-
- public void setImplicitCommitSize(Integer implicitCommitSize) {
- this.implicitCommitSize = implicitCommitSize;
- }
-
public void enqueue(ClusterAction action) {
actionCount++;
this.actions.add(action);
- if (this.actions.size() > getImplicitCommitSize()) {
+ if (this.actions.size() > IMPLICIT_COMMIT_SIZE) {
this.commit();
}
}
projects / sonarqube.git / commitdiff