Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Follow stats structure #34301

Merged
merged 5 commits into from
Oct 5, 2018
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
---
"Test stats":
- do:
indices.create:
index: foo
body:
settings:
index:
soft_deletes:
enabled: true
mappings:
doc:
properties:
field:
type: keyword

- do:
ccr.follow:
index: bar
body:
leader_index: foo
- is_true: follow_index_created
- is_true: follow_index_shards_acked
- is_true: index_following_started

# we can not reliably wait for replication to occur so we test the endpoint without indexing any documents
- do:
ccr.stats:
index: bar
- match: { indices.0.index: "bar" }
- match: { indices.0.shards.0.leader_index: "foo" }
- match: { indices.0.shards.0.follower_index: "bar" }
- match: { indices.0.shards.0.shard_id: 0 }
- gte: { indices.0.shards.0.leader_global_checkpoint: -1 }
- gte: { indices.0.shards.0.leader_max_seq_no: -1 }
- gte: { indices.0.shards.0.follower_global_checkpoint: -1 }
- gte: { indices.0.shards.0.follower_max_seq_no: -1 }
- gte: { indices.0.shards.0.last_requested_seq_no: -1 }
- gte: { indices.0.shards.0.number_of_concurrent_reads: 0 }
- match: { indices.0.shards.0.number_of_concurrent_writes: 0 }
- match: { indices.0.shards.0.number_of_queued_writes: 0 }
- gte: { indices.0.shards.0.mapping_version: 0 }
- gte: { indices.0.shards.0.total_fetch_time_millis: 0 }
- gte: { indices.0.shards.0.number_of_successful_fetches: 0 }
- gte: { indices.0.shards.0.number_of_failed_fetches: 0 }
- match: { indices.0.shards.0.operations_received: 0 }
- match: { indices.0.shards.0.total_transferred_bytes: 0 }
- match: { indices.0.shards.0.total_index_time_millis: 0 }
- match: { indices.0.shards.0.number_of_successful_bulk_operations: 0 }
- match: { indices.0.shards.0.number_of_failed_bulk_operations: 0 }
- match: { indices.0.shards.0.number_of_operations_indexed: 0 }
- length: { indices.0.shards.0.fetch_exceptions: 0 }
- gte: { indices.0.shards.0.time_since_last_fetch_millis: -1 }

- do:
ccr.pause_follow:
index: bar
- is_true: acknowledged

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@
import org.elasticsearch.xpack.ccr.action.ShardChangesAction;
import org.elasticsearch.xpack.ccr.action.ShardFollowTask;
import org.elasticsearch.xpack.ccr.action.ShardFollowTasksExecutor;
import org.elasticsearch.xpack.ccr.action.TransportCcrStatsAction;
import org.elasticsearch.xpack.ccr.action.TransportFollowStatsAction;
import org.elasticsearch.xpack.ccr.action.TransportPutFollowAction;
import org.elasticsearch.xpack.ccr.action.TransportDeleteAutoFollowPatternAction;
import org.elasticsearch.xpack.ccr.action.TransportResumeFollowAction;
Expand All @@ -62,15 +62,15 @@
import org.elasticsearch.xpack.ccr.action.bulk.BulkShardOperationsAction;
import org.elasticsearch.xpack.ccr.action.bulk.TransportBulkShardOperationsAction;
import org.elasticsearch.xpack.ccr.index.engine.FollowingEngineFactory;
import org.elasticsearch.xpack.ccr.rest.RestCcrStatsAction;
import org.elasticsearch.xpack.ccr.rest.RestFollowStatsAction;
import org.elasticsearch.xpack.ccr.rest.RestPutFollowAction;
import org.elasticsearch.xpack.ccr.rest.RestDeleteAutoFollowPatternAction;
import org.elasticsearch.xpack.ccr.rest.RestResumeFollowAction;
import org.elasticsearch.xpack.ccr.rest.RestPutAutoFollowPatternAction;
import org.elasticsearch.xpack.ccr.rest.RestPauseFollowAction;
import org.elasticsearch.xpack.core.XPackPlugin;
import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;
import org.elasticsearch.xpack.core.ccr.action.PutFollowAction;
import org.elasticsearch.xpack.core.ccr.action.ResumeFollowAction;
import org.elasticsearch.xpack.core.ccr.action.PauseFollowAction;
Expand Down Expand Up @@ -161,7 +161,7 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(ClusterServic
new ActionHandler<>(BulkShardOperationsAction.INSTANCE, TransportBulkShardOperationsAction.class),
new ActionHandler<>(ShardChangesAction.INSTANCE, ShardChangesAction.TransportAction.class),
// stats action
new ActionHandler<>(CcrStatsAction.INSTANCE, TransportCcrStatsAction.class),
new ActionHandler<>(FollowStatsAction.INSTANCE, TransportFollowStatsAction.class),
new ActionHandler<>(AutoFollowStatsAction.INSTANCE, TransportAutoFollowStatsAction.class),
// follow actions
new ActionHandler<>(PutFollowAction.INSTANCE, TransportPutFollowAction.class),
Expand All @@ -184,7 +184,7 @@ public List<RestHandler> getRestHandlers(Settings settings, RestController restC

return Arrays.asList(
// stats API
new RestCcrStatsAction(settings, restController),
new RestFollowStatsAction(settings, restController),
new RestAutoFollowStatsAction(settings, restController),
// follow APIs
new RestPutFollowAction(settings, restController),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.xpack.ccr.Ccr;
import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;

import java.io.IOException;
import java.util.Arrays;
Expand All @@ -32,16 +32,16 @@
import java.util.Set;
import java.util.function.Consumer;

public class TransportCcrStatsAction extends TransportTasksAction<
public class TransportFollowStatsAction extends TransportTasksAction<
ShardFollowNodeTask,
CcrStatsAction.StatsRequest,
CcrStatsAction.StatsResponses, CcrStatsAction.StatsResponse> {
FollowStatsAction.StatsRequest,
FollowStatsAction.StatsResponses, FollowStatsAction.StatsResponse> {

private final IndexNameExpressionResolver resolver;
private final CcrLicenseChecker ccrLicenseChecker;

@Inject
public TransportCcrStatsAction(
public TransportFollowStatsAction(
final Settings settings,
final ClusterService clusterService,
final TransportService transportService,
Expand All @@ -50,12 +50,12 @@ public TransportCcrStatsAction(
final CcrLicenseChecker ccrLicenseChecker) {
super(
settings,
CcrStatsAction.NAME,
FollowStatsAction.NAME,
clusterService,
transportService,
actionFilters,
CcrStatsAction.StatsRequest::new,
CcrStatsAction.StatsResponses::new,
FollowStatsAction.StatsRequest::new,
FollowStatsAction.StatsResponses::new,
Ccr.CCR_THREAD_POOL_NAME);
this.resolver = Objects.requireNonNull(resolver);
this.ccrLicenseChecker = Objects.requireNonNull(ccrLicenseChecker);
Expand All @@ -64,8 +64,8 @@ public TransportCcrStatsAction(
@Override
protected void doExecute(
final Task task,
final CcrStatsAction.StatsRequest request,
final ActionListener<CcrStatsAction.StatsResponses> listener) {
final FollowStatsAction.StatsRequest request,
final ActionListener<FollowStatsAction.StatsResponses> listener) {
if (ccrLicenseChecker.isCcrAllowed() == false) {
listener.onFailure(LicenseUtils.newComplianceException("ccr"));
return;
Expand All @@ -74,21 +74,21 @@ protected void doExecute(
}

@Override
protected CcrStatsAction.StatsResponses newResponse(
final CcrStatsAction.StatsRequest request,
final List<CcrStatsAction.StatsResponse> statsRespons,
protected FollowStatsAction.StatsResponses newResponse(
final FollowStatsAction.StatsRequest request,
final List<FollowStatsAction.StatsResponse> statsRespons,
final List<TaskOperationFailure> taskOperationFailures,
final List<FailedNodeException> failedNodeExceptions) {
return new CcrStatsAction.StatsResponses(taskOperationFailures, failedNodeExceptions, statsRespons);
return new FollowStatsAction.StatsResponses(taskOperationFailures, failedNodeExceptions, statsRespons);
}

@Override
protected CcrStatsAction.StatsResponse readTaskResponse(final StreamInput in) throws IOException {
return new CcrStatsAction.StatsResponse(in);
protected FollowStatsAction.StatsResponse readTaskResponse(final StreamInput in) throws IOException {
return new FollowStatsAction.StatsResponse(in);
}

@Override
protected void processTasks(final CcrStatsAction.StatsRequest request, final Consumer<ShardFollowNodeTask> operation) {
protected void processTasks(final FollowStatsAction.StatsRequest request, final Consumer<ShardFollowNodeTask> operation) {
final ClusterState state = clusterService.state();
final Set<String> concreteIndices = new HashSet<>(Arrays.asList(resolver.concreteIndexNames(state, request)));
for (final Task task : taskManager.getTasks().values()) {
Expand All @@ -103,10 +103,10 @@ protected void processTasks(final CcrStatsAction.StatsRequest request, final Con

@Override
protected void taskOperation(
final CcrStatsAction.StatsRequest request,
final FollowStatsAction.StatsRequest request,
final ShardFollowNodeTask task,
final ActionListener<CcrStatsAction.StatsResponse> listener) {
listener.onResponse(new CcrStatsAction.StatsResponse(task.getStatus()));
final ActionListener<FollowStatsAction.StatsResponse> listener) {
listener.onResponse(new FollowStatsAction.StatsResponse(task.getStatus()));
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,13 @@
import org.elasticsearch.rest.RestController;
import org.elasticsearch.rest.RestRequest;
import org.elasticsearch.rest.action.RestToXContentListener;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;

import java.io.IOException;

public class RestCcrStatsAction extends BaseRestHandler {
public class RestFollowStatsAction extends BaseRestHandler {

public RestCcrStatsAction(final Settings settings, final RestController controller) {
public RestFollowStatsAction(final Settings settings, final RestController controller) {
super(settings);
controller.registerHandler(RestRequest.Method.GET, "/_ccr/stats", this);
controller.registerHandler(RestRequest.Method.GET, "/{index}/_ccr/stats", this);
Expand All @@ -32,9 +32,9 @@ public String getName() {

@Override
protected RestChannelConsumer prepareRequest(final RestRequest restRequest, final NodeClient client) throws IOException {
final CcrStatsAction.StatsRequest request = new CcrStatsAction.StatsRequest();
final FollowStatsAction.StatsRequest request = new FollowStatsAction.StatsRequest();
request.setIndices(Strings.splitStringByCommaToArray(restRequest.param("index")));
return channel -> client.execute(CcrStatsAction.INSTANCE, request, new RestToXContentListener<>(channel));
return channel -> client.execute(FollowStatsAction.INSTANCE, request, new RestToXContentListener<>(channel));
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import org.elasticsearch.test.ESSingleNodeTestCase;
import org.elasticsearch.test.MockLogAppender;
import org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;
import org.elasticsearch.xpack.core.ccr.action.PutFollowAction;
import org.elasticsearch.xpack.core.ccr.action.ResumeFollowAction;
import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata;
Expand Down Expand Up @@ -88,21 +88,24 @@ public void onFailure(final Exception e) {
latch.await();
}

public void testThatCcrStatsAreUnavailableWithNonCompliantLicense() throws InterruptedException {
public void testThatFollowStatsAreUnavailableWithNonCompliantLicense() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
client().execute(CcrStatsAction.INSTANCE, new CcrStatsAction.StatsRequest(), new ActionListener<CcrStatsAction.StatsResponses>() {
@Override
public void onResponse(final CcrStatsAction.StatsResponses statsResponses) {
latch.countDown();
fail();
}
client().execute(
FollowStatsAction.INSTANCE,
new FollowStatsAction.StatsRequest(),
new ActionListener<FollowStatsAction.StatsResponses>() {
@Override
public void onResponse(final FollowStatsAction.StatsResponses statsResponses) {
latch.countDown();
fail();
}

@Override
public void onFailure(final Exception e) {
assertNonCompliantLicense(e);
latch.countDown();
}
});
@Override
public void onFailure(final Exception e) {
assertNonCompliantLicense(e);
latch.countDown();
}
});

latch.await();
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,9 +59,9 @@
import org.elasticsearch.xpack.ccr.index.engine.FollowingEngine;
import org.elasticsearch.xpack.core.XPackSettings;
import org.elasticsearch.xpack.core.ccr.ShardFollowNodeTaskStatus;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction.StatsRequest;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction.StatsResponses;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction.StatsRequest;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction.StatsResponses;
import org.elasticsearch.xpack.core.ccr.action.PutFollowAction;
import org.elasticsearch.xpack.core.ccr.action.ResumeFollowAction;
import org.elasticsearch.xpack.core.ccr.action.PauseFollowAction;
Expand Down Expand Up @@ -570,7 +570,7 @@ public void testCloseLeaderIndex() throws Exception {

client().admin().indices().close(new CloseIndexRequest("index1")).actionGet();
assertBusy(() -> {
StatsResponses response = client().execute(CcrStatsAction.INSTANCE, new StatsRequest()).actionGet();
StatsResponses response = client().execute(FollowStatsAction.INSTANCE, new StatsRequest()).actionGet();
assertThat(response.getNodeFailures(), empty());
assertThat(response.getTaskFailures(), empty());
assertThat(response.getStatsResponses(), hasSize(1));
Expand Down Expand Up @@ -605,7 +605,7 @@ public void testCloseFollowIndex() throws Exception {
client().admin().indices().close(new CloseIndexRequest("index2")).actionGet();
client().prepareIndex("index1", "doc", "2").setSource("{}", XContentType.JSON).get();
assertBusy(() -> {
StatsResponses response = client().execute(CcrStatsAction.INSTANCE, new StatsRequest()).actionGet();
StatsResponses response = client().execute(FollowStatsAction.INSTANCE, new StatsRequest()).actionGet();
assertThat(response.getNodeFailures(), empty());
assertThat(response.getTaskFailures(), empty());
assertThat(response.getStatsResponses(), hasSize(1));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,18 +6,18 @@
package org.elasticsearch.xpack.ccr.action;

import org.elasticsearch.test.AbstractStreamableTestCase;
import org.elasticsearch.xpack.core.ccr.action.CcrStatsAction;
import org.elasticsearch.xpack.core.ccr.action.FollowStatsAction;

public class StatsRequestTests extends AbstractStreamableTestCase<CcrStatsAction.StatsRequest> {
public class StatsRequestTests extends AbstractStreamableTestCase<FollowStatsAction.StatsRequest> {

@Override
protected CcrStatsAction.StatsRequest createBlankInstance() {
return new CcrStatsAction.StatsRequest();
protected FollowStatsAction.StatsRequest createBlankInstance() {
return new FollowStatsAction.StatsRequest();
}

@Override
protected CcrStatsAction.StatsRequest createTestInstance() {
CcrStatsAction.StatsRequest statsRequest = new CcrStatsAction.StatsRequest();
protected FollowStatsAction.StatsRequest createTestInstance() {
FollowStatsAction.StatsRequest statsRequest = new FollowStatsAction.StatsRequest();
if (randomBoolean()) {
statsRequest.setIndices(generateRandomStringArray(8, 4, false));
}
Expand Down
Loading