Skip to content

Commit f060509

Browse files
jbaierajbaiera
authored
Refactor IngestService node info collection (#132656)
This refactors the node info collection logic to be injected into the IngestService constructor instead of requiring it to be specified on every call to put templates. This makes the logic easier to reuse in the future.
1 parent 3f9a248 commit f060509

File tree

3 files changed

+70
-40
lines changed

3 files changed

+70
-40
lines changed

‎server/src/main/java/org/elasticsearch/action/ingest/PutPipelineTransportAction.java‎

Lines changed: 2 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,9 @@
1111

1212
import org.elasticsearch.action.ActionListener;
1313
import org.elasticsearch.action.ActionType;
14-
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoMetrics;
15-
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
1614
import org.elasticsearch.action.support.ActionFilters;
1715
import org.elasticsearch.action.support.master.AcknowledgedResponse;
1816
import org.elasticsearch.action.support.master.AcknowledgedTransportMasterNodeAction;
19-
import org.elasticsearch.client.internal.OriginSettingClient;
20-
import org.elasticsearch.client.internal.node.NodeClient;
2117
import org.elasticsearch.cluster.ClusterState;
2218
import org.elasticsearch.cluster.block.ClusterBlockException;
2319
import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -32,12 +28,9 @@
3228
import java.util.Optional;
3329
import java.util.Set;
3430

35-
import static org.elasticsearch.ingest.IngestService.INGEST_ORIGIN;
36-
3731
public class PutPipelineTransportAction extends AcknowledgedTransportMasterNodeAction<PutPipelineRequest> {
3832
public static final ActionType<AcknowledgedResponse> TYPE = new ActionType<>("cluster:admin/ingest/pipeline/put");
3933
private final IngestService ingestService;
40-
private final OriginSettingClient client;
4134
private final ProjectResolver projectResolver;
4235

4336
@Inject
@@ -46,8 +39,7 @@ public PutPipelineTransportAction(
4639
TransportService transportService,
4740
ActionFilters actionFilters,
4841
ProjectResolver projectResolver,
49-
IngestService ingestService,
50-
NodeClient client
42+
IngestService ingestService
5143
) {
5244
super(
5345
TYPE.name(),
@@ -58,22 +50,14 @@ public PutPipelineTransportAction(
5850
PutPipelineRequest::new,
5951
EsExecutors.DIRECT_EXECUTOR_SERVICE
6052
);
61-
// This client is only used to perform an internal implementation detail,
62-
// so uses an internal origin context rather than the user context
63-
this.client = new OriginSettingClient(client, INGEST_ORIGIN);
6453
this.ingestService = ingestService;
6554
this.projectResolver = projectResolver;
6655
}
6756

6857
@Override
6958
protected void masterOperation(Task task, PutPipelineRequest request, ClusterState state, ActionListener<AcknowledgedResponse> listener)
7059
throws Exception {
71-
ingestService.putPipeline(projectResolver.getProjectId(), request, listener, (nodeListener) -> {
72-
NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
73-
nodesInfoRequest.clear();
74-
nodesInfoRequest.addMetric(NodesInfoMetrics.Metric.INGEST.metricName());
75-
client.admin().cluster().nodesInfo(nodesInfoRequest, nodeListener);
76-
});
60+
ingestService.putPipeline(projectResolver.getProjectId(), request, listener);
7761
}
7862

7963
@Override

‎server/src/main/java/org/elasticsearch/ingest/IngestService.java‎

Lines changed: 51 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919
import org.elasticsearch.action.ActionListener;
2020
import org.elasticsearch.action.DocWriteRequest;
2121
import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
22+
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoMetrics;
23+
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
2224
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
2325
import org.elasticsearch.action.bulk.FailureStoreMetrics;
2426
import org.elasticsearch.action.bulk.IndexDocFailureStoreStatus;
@@ -29,6 +31,7 @@
2931
import org.elasticsearch.action.support.RefCountingRunnable;
3032
import org.elasticsearch.action.support.master.AcknowledgedResponse;
3133
import org.elasticsearch.client.internal.Client;
34+
import org.elasticsearch.client.internal.OriginSettingClient;
3235
import org.elasticsearch.cluster.ClusterChangedEvent;
3336
import org.elasticsearch.cluster.ClusterState;
3437
import org.elasticsearch.cluster.ClusterStateApplier;
@@ -153,11 +156,24 @@ public static boolean locallySupportedIngestFeature(NodeFeature nodeFeature) {
153156
private volatile ClusterState state;
154157
private final ProjectResolver projectResolver;
155158
private final FeatureService featureService;
159+
private final Consumer<ActionListener<NodesInfoResponse>> nodeInfoListener;
156160

157161
private static BiFunction<Long, Runnable, Scheduler.ScheduledCancellable> createScheduler(ThreadPool threadPool) {
158162
return (delay, command) -> threadPool.schedule(command, TimeValue.timeValueMillis(delay), threadPool.generic());
159163
}
160164

165+
private static Consumer<ActionListener<NodesInfoResponse>> createNodeInfoListener(Client client) {
166+
// This client is only used to perform an internal implementation detail,
167+
// so uses an internal origin context rather than the user context
168+
final OriginSettingClient originSettingClient = new OriginSettingClient(client, INGEST_ORIGIN);
169+
return (nodeListener) -> {
170+
NodesInfoRequest nodesInfoRequest = new NodesInfoRequest();
171+
nodesInfoRequest.clear();
172+
nodesInfoRequest.addMetric(NodesInfoMetrics.Metric.INGEST.metricName());
173+
originSettingClient.admin().cluster().nodesInfo(nodesInfoRequest, nodeListener);
174+
};
175+
}
176+
161177
public static MatcherWatchdog createGrokThreadWatchdog(Environment env, ThreadPool threadPool) {
162178
final Settings settings = env.settings();
163179
final BiFunction<Long, Runnable, Scheduler.ScheduledCancellable> scheduler = createScheduler(threadPool);
@@ -240,7 +256,8 @@ public IngestService(
240256
MatcherWatchdog matcherWatchdog,
241257
FailureStoreMetrics failureStoreMetrics,
242258
ProjectResolver projectResolver,
243-
FeatureService featureService
259+
FeatureService featureService,
260+
Consumer<ActionListener<NodesInfoResponse>> nodeInfoListener
244261
) {
245262
this.clusterService = clusterService;
246263
this.scriptService = scriptService;
@@ -264,6 +281,36 @@ public IngestService(
264281
this.failureStoreMetrics = failureStoreMetrics;
265282
this.projectResolver = projectResolver;
266283
this.featureService = featureService;
284+
this.nodeInfoListener = nodeInfoListener;
285+
}
286+
287+
public IngestService(
288+
ClusterService clusterService,
289+
ThreadPool threadPool,
290+
Environment env,
291+
ScriptService scriptService,
292+
AnalysisRegistry analysisRegistry,
293+
List<IngestPlugin> ingestPlugins,
294+
Client client,
295+
MatcherWatchdog matcherWatchdog,
296+
FailureStoreMetrics failureStoreMetrics,
297+
ProjectResolver projectResolver,
298+
FeatureService featureService
299+
) {
300+
this(
301+
clusterService,
302+
threadPool,
303+
env,
304+
scriptService,
305+
analysisRegistry,
306+
ingestPlugins,
307+
client,
308+
matcherWatchdog,
309+
failureStoreMetrics,
310+
projectResolver,
311+
featureService,
312+
createNodeInfoListener(client)
313+
);
267314
}
268315

269316
/**
@@ -282,6 +329,7 @@ public IngestService(
282329
this.failureStoreMetrics = ingestService.failureStoreMetrics;
283330
this.projectResolver = ingestService.projectResolver;
284331
this.featureService = ingestService.featureService;
332+
this.nodeInfoListener = ingestService.nodeInfoListener;
285333
}
286334

287335
private static Map<String, Processor.Factory> processorFactories(List<IngestPlugin> ingestPlugins, Processor.Parameters parameters) {
@@ -535,12 +583,8 @@ static List<PipelineConfiguration> innerGetPipelines(IngestMetadata ingestMetada
535583
/**
536584
* Stores the specified pipeline definition in the request.
537585
*/
538-
public void putPipeline(
539-
ProjectId projectId,
540-
PutPipelineRequest request,
541-
ActionListener<AcknowledgedResponse> listener,
542-
Consumer<ActionListener<NodesInfoResponse>> nodeInfoListener
543-
) throws Exception {
586+
public void putPipeline(ProjectId projectId, PutPipelineRequest request, ActionListener<AcknowledgedResponse> listener)
587+
throws Exception {
544588
if (isNoOpPipelineUpdate(state.metadata().getProject(projectId), request)) {
545589
// existing pipeline matches request pipeline -- no need to update
546590
listener.onResponse(AcknowledgedResponse.TRUE);

‎server/src/test/java/org/elasticsearch/ingest/IngestServiceTests.java‎

Lines changed: 17 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -3044,6 +3044,20 @@ private void testUpdatingPipeline(String pipelineString) throws Exception {
30443044
Client client = mock(Client.class);
30453045
ClusterService clusterService = mock(ClusterService.class);
30463046
when(clusterService.state()).thenReturn(clusterState);
3047+
3048+
var consumer = new Consumer<ActionListener<NodesInfoResponse>>() {
3049+
final AtomicLong executionCount = new AtomicLong(0);
3050+
3051+
@Override
3052+
public void accept(ActionListener<NodesInfoResponse> nodesInfoResponseActionListener) {
3053+
executionCount.incrementAndGet();
3054+
}
3055+
3056+
public long getExecutionCount() {
3057+
return executionCount.get();
3058+
}
3059+
};
3060+
30473061
IngestService ingestService = new IngestService(
30483062
clusterService,
30493063
threadPool,
@@ -3060,7 +3074,8 @@ private void testUpdatingPipeline(String pipelineString) throws Exception {
30603074
public boolean clusterHasFeature(ClusterState state, NodeFeature feature) {
30613075
return DataStream.DATA_STREAM_FAILURE_STORE_FEATURE.equals(feature);
30623076
}
3063-
}
3077+
},
3078+
consumer
30643079
);
30653080
ingestService.applyClusterState(new ClusterChangedEvent("", clusterState, clusterState));
30663081

@@ -3090,21 +3105,8 @@ public long getFailureCount() {
30903105
}
30913106
};
30923107

3093-
var consumer = new Consumer<ActionListener<NodesInfoResponse>>() {
3094-
final AtomicLong executionCount = new AtomicLong(0);
3095-
3096-
@Override
3097-
public void accept(ActionListener<NodesInfoResponse> nodesInfoResponseActionListener) {
3098-
executionCount.incrementAndGet();
3099-
}
3100-
3101-
public long getExecutionCount() {
3102-
return executionCount.get();
3103-
}
3104-
};
3105-
31063108
var request = putJsonPipelineRequest(pipelineId, pipelineString);
3107-
ingestService.putPipeline(clusterState.metadata().getProject().id(), request, listener, consumer);
3109+
ingestService.putPipeline(clusterState.metadata().getProject().id(), request, listener);
31083110
latch.await();
31093111

31103112
assertThat(consumer.getExecutionCount(), equalTo(0L));

0 commit comments

Comments
 (0)