Skip to content

Commit 3200abc

Browse files
samxbrsamxbr
authored
Make EnterpriseGeoIpDownloaderLicenseListener project aware (#129992)
1 parent 93e4e01 commit 3200abc

File tree

4 files changed

+90
-41
lines changed

4 files changed

+90
-41
lines changed

‎server/src/main/java/org/elasticsearch/persistent/PersistentTasksService.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -296,6 +296,7 @@ private void sendUpdateStateRequest(
296296
/**
297297
* Notifies the master node to remove a persistent task from the cluster state. Accepts operation timeout as optional parameter
298298
*/
299+
@Deprecated(forRemoval = true) // Use the explict cluster/project version instead
299300
public void sendRemoveRequest(final String taskId, final TimeValue timeout, final ActionListener<PersistentTask<?>> listener) {
300301
sendRemoveRequest(null, taskId, timeout, listener);
301302
}

‎x-pack/plugin/geoip-enterprise-downloader/src/main/java/org/elasticsearch/xpack/geoip/EnterpriseDownloaderPlugin.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,8 @@ public Collection<?> createComponents(PluginServices services) {
4040
services.client(),
4141
services.clusterService(),
4242
services.threadPool(),
43-
getLicenseState()
43+
getLicenseState(),
44+
services.projectResolver()
4445
);
4546
enterpriseGeoIpDownloaderLicenseListener.init();
4647
return List.of(enterpriseGeoIpDownloaderLicenseListener);

‎x-pack/plugin/geoip-enterprise-downloader/src/main/java/org/elasticsearch/xpack/geoip/EnterpriseGeoIpDownloaderLicenseListener.java‎

Lines changed: 53 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,11 @@
1616
import org.elasticsearch.action.support.master.MasterNodeRequest;
1717
import org.elasticsearch.client.internal.Client;
1818
import org.elasticsearch.cluster.ClusterChangedEvent;
19-
import org.elasticsearch.cluster.ClusterState;
2019
import org.elasticsearch.cluster.ClusterStateListener;
20+
import org.elasticsearch.cluster.metadata.ProjectId;
21+
import org.elasticsearch.cluster.project.ProjectResolver;
2122
import org.elasticsearch.cluster.service.ClusterService;
23+
import org.elasticsearch.core.NotMultiProjectCapable;
2224
import org.elasticsearch.ingest.EnterpriseGeoIpTask.EnterpriseGeoIpTaskParams;
2325
import org.elasticsearch.license.License;
2426
import org.elasticsearch.license.LicenseStateListener;
@@ -31,12 +33,14 @@
3133
import org.elasticsearch.xpack.core.XPackField;
3234

3335
import java.util.Objects;
36+
import java.util.concurrent.ConcurrentHashMap;
37+
import java.util.concurrent.ConcurrentMap;
3438

3539
import static org.elasticsearch.ingest.EnterpriseGeoIpTask.ENTERPRISE_GEOIP_DOWNLOADER;
3640

3741
public class EnterpriseGeoIpDownloaderLicenseListener implements LicenseStateListener, ClusterStateListener {
3842
private static final Logger logger = LogManager.getLogger(EnterpriseGeoIpDownloaderLicenseListener.class);
39-
// Note: This custom type is GeoIpMetadata.TYPE, but that class is not exposed to this plugin
43+
// Note: This custom type is IngestGeoIpMetadata.TYPE, but that class is not exposed to this plugin
4044
static final String INGEST_GEOIP_CUSTOM_METADATA_TYPE = "ingest_geoip";
4145

4246
private final PersistentTasksService persistentTasksService;
@@ -47,18 +51,21 @@ public class EnterpriseGeoIpDownloaderLicenseListener implements LicenseStateLis
4751
XPackField.ENTERPRISE_GEOIP_DOWNLOADER,
4852
License.OperationMode.PLATINUM
4953
);
50-
private volatile boolean licenseIsValid = false;
51-
private volatile boolean hasIngestGeoIpMetadata = false;
54+
private final ConcurrentMap<ProjectId, Boolean> licenseIsValid = new ConcurrentHashMap<>();
55+
private final ConcurrentMap<ProjectId, Boolean> hasIngestGeoIpMetadata = new ConcurrentHashMap<>();
56+
private final ProjectResolver projectResolver;
5257

5358
protected EnterpriseGeoIpDownloaderLicenseListener(
5459
Client client,
5560
ClusterService clusterService,
5661
ThreadPool threadPool,
57-
XPackLicenseState licenseState
62+
XPackLicenseState licenseState,
63+
ProjectResolver projectResolver
5864
) {
5965
this.persistentTasksService = new PersistentTasksService(clusterService, threadPool, client);
6066
this.clusterService = clusterService;
6167
this.licenseState = licenseState;
68+
this.projectResolver = projectResolver;
6269
}
6370

6471
private volatile boolean licenseStateListenerRegistered;
@@ -74,47 +81,55 @@ void listenForLicenseStateChanges() {
7481
licenseState.addListener(this);
7582
}
7683

84+
@NotMultiProjectCapable(description = "Replace DEFAULT project after enterprise license is supported in serverless and project-aware")
7785
@Override
7886
public void licenseStateChanged() {
79-
licenseIsValid = ENTERPRISE_GEOIP_FEATURE.checkWithoutTracking(licenseState);
80-
maybeUpdateTaskState(clusterService.state());
87+
licenseIsValid.put(ProjectId.DEFAULT, ENTERPRISE_GEOIP_FEATURE.checkWithoutTracking(licenseState));
88+
final boolean isLocalNodeMaster = clusterService.state().nodes().isLocalNodeElectedMaster();
89+
maybeUpdateTaskState(ProjectId.DEFAULT, isLocalNodeMaster);
8190
}
8291

8392
@Override
8493
public void clusterChanged(ClusterChangedEvent event) {
85-
hasIngestGeoIpMetadata = event.state().metadata().getProject().custom(INGEST_GEOIP_CUSTOM_METADATA_TYPE) != null;
86-
final boolean ingestGeoIpCustomMetaChangedInEvent = event.metadataChanged()
87-
&& event.changedCustomProjectMetadataSet().contains(INGEST_GEOIP_CUSTOM_METADATA_TYPE);
8894
final boolean masterNodeChanged = Objects.equals(
8995
event.state().nodes().getMasterNode(),
9096
event.previousState().nodes().getMasterNode()
9197
) == false;
92-
/*
93-
* We don't want to potentially start the task on every cluster state change, so only maybeUpdateTaskState if this cluster change
94-
* event involved the modification of custom geoip metadata OR a master node change
95-
*/
96-
if (ingestGeoIpCustomMetaChangedInEvent || (masterNodeChanged && hasIngestGeoIpMetadata)) {
97-
maybeUpdateTaskState(event.state());
98-
}
98+
final boolean isLocalNodeMaster = event.state().nodes().isLocalNodeElectedMaster();
99+
event.state().metadata().projects().values().forEach(projectMetadata -> {
100+
ProjectId projectId = projectMetadata.id();
101+
final boolean hasMetadata = projectMetadata.custom(INGEST_GEOIP_CUSTOM_METADATA_TYPE) != null;
102+
hasIngestGeoIpMetadata.put(projectId, hasMetadata);
103+
final boolean ingestGeoIpCustomMetaChangedInEvent = event.metadataChanged()
104+
&& event.customMetadataChanged(projectId, INGEST_GEOIP_CUSTOM_METADATA_TYPE);
105+
/*
106+
* We don't want to potentially start the task on every cluster state change, so only maybeUpdateTaskState
107+
* if this cluster change event involved the modification of custom geoip metadata OR a master node change
108+
*/
109+
if (ingestGeoIpCustomMetaChangedInEvent || (masterNodeChanged && hasIngestGeoIpMetadata.getOrDefault(projectId, false))) {
110+
maybeUpdateTaskState(projectId, isLocalNodeMaster);
111+
}
112+
});
99113
}
100114

101-
private void maybeUpdateTaskState(ClusterState state) {
115+
private void maybeUpdateTaskState(ProjectId projectId, boolean isLocalNodeMaster) {
102116
// We should only start/stop task from single node, master is the best as it will go through it anyway
103-
if (state.nodes().isLocalNodeElectedMaster()) {
104-
if (licenseIsValid) {
105-
if (hasIngestGeoIpMetadata) {
106-
ensureTaskStarted();
117+
if (isLocalNodeMaster) {
118+
if (licenseIsValid.getOrDefault(projectId, false)) {
119+
if (hasIngestGeoIpMetadata.getOrDefault(projectId, false)) {
120+
ensureTaskStarted(projectId);
107121
}
108122
} else {
109-
ensureTaskStopped();
123+
ensureTaskStopped(projectId);
110124
}
111125
}
112126
}
113127

114-
private void ensureTaskStarted() {
115-
assert licenseIsValid : "Task should never be started without valid license";
116-
persistentTasksService.sendStartRequest(
117-
ENTERPRISE_GEOIP_DOWNLOADER,
128+
private void ensureTaskStarted(ProjectId projectId) {
129+
assert licenseIsValid.getOrDefault(projectId, false) : "Task should never be started without valid license";
130+
persistentTasksService.sendProjectStartRequest(
131+
projectId,
132+
getTaskId(projectId, projectResolver.supportsMultipleProjects()),
118133
ENTERPRISE_GEOIP_DOWNLOADER,
119134
new EnterpriseGeoIpTaskParams(),
120135
MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
@@ -127,7 +142,7 @@ private void ensureTaskStarted() {
127142
);
128143
}
129144

130-
private void ensureTaskStopped() {
145+
private void ensureTaskStopped(ProjectId projectId) {
131146
ActionListener<PersistentTasksCustomMetadata.PersistentTask<?>> listener = ActionListener.wrap(
132147
r -> logger.debug("Stopped enterprise geoip downloader task"),
133148
e -> {
@@ -137,6 +152,15 @@ private void ensureTaskStopped() {
137152
}
138153
}
139154
);
140-
persistentTasksService.sendRemoveRequest(ENTERPRISE_GEOIP_DOWNLOADER, MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT, listener);
155+
persistentTasksService.sendProjectRemoveRequest(
156+
projectId,
157+
getTaskId(projectId, projectResolver.supportsMultipleProjects()),
158+
MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
159+
listener
160+
);
161+
}
162+
163+
protected static String getTaskId(ProjectId projectId, boolean supportsMultipleProjects) {
164+
return supportsMultipleProjects ? projectId + "/" + ENTERPRISE_GEOIP_DOWNLOADER : ENTERPRISE_GEOIP_DOWNLOADER;
141165
}
142166
}

‎x-pack/plugin/geoip-enterprise-downloader/src/test/java/org/elasticsearch/xpack/geoip/EnterpriseGeoIpDownloaderLicenseListenerTests.java‎

Lines changed: 34 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,14 @@
1616
import org.elasticsearch.cluster.ClusterState;
1717
import org.elasticsearch.cluster.metadata.IndexMetadata;
1818
import org.elasticsearch.cluster.metadata.Metadata;
19+
import org.elasticsearch.cluster.metadata.ProjectMetadata;
1920
import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
2021
import org.elasticsearch.cluster.node.DiscoveryNodes;
22+
import org.elasticsearch.cluster.project.ProjectResolver;
23+
import org.elasticsearch.cluster.project.TestProjectResolvers;
2124
import org.elasticsearch.cluster.service.ClusterService;
2225
import org.elasticsearch.common.settings.Settings;
26+
import org.elasticsearch.core.NotMultiProjectCapable;
2327
import org.elasticsearch.index.Index;
2428
import org.elasticsearch.index.IndexVersion;
2529
import org.elasticsearch.license.License;
@@ -48,6 +52,8 @@
4852
public class EnterpriseGeoIpDownloaderLicenseListenerTests extends ESTestCase {
4953

5054
private ThreadPool threadPool;
55+
@NotMultiProjectCapable(description = "Enterprise license not available in serverless or multi-project yet")
56+
private final ProjectResolver projectResolver = TestProjectResolvers.DEFAULT_PROJECT_ONLY;
5157

5258
@Before
5359
public void setup() {
@@ -68,12 +74,13 @@ public void testAllConditionsMetOnStart() {
6874
// Should never start if not master node, even if all other conditions have been met
6975
final XPackLicenseState licenseState = getAlwaysValidLicense();
7076
ClusterService clusterService = createClusterService(true, false);
71-
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, true, false);
77+
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, true, false, projectResolver);
7278
EnterpriseGeoIpDownloaderLicenseListener listener = new EnterpriseGeoIpDownloaderLicenseListener(
7379
client,
7480
clusterService,
7581
threadPool,
76-
licenseState
82+
licenseState,
83+
projectResolver
7784
);
7885
listener.init();
7986
listener.licenseStateChanged();
@@ -85,12 +92,13 @@ public void testLicenseChanges() {
8592
final TestUtils.UpdatableLicenseState licenseState = new TestUtils.UpdatableLicenseState();
8693
licenseState.update(new XPackLicenseStatus(License.OperationMode.TRIAL, false, ""));
8794
ClusterService clusterService = createClusterService(true, true);
88-
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, false, true);
95+
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, false, true, projectResolver);
8996
EnterpriseGeoIpDownloaderLicenseListener listener = new EnterpriseGeoIpDownloaderLicenseListener(
9097
client,
9198
clusterService,
9299
threadPool,
93-
licenseState
100+
licenseState,
101+
projectResolver
94102
);
95103
listener.init();
96104
listener.licenseStateChanged();
@@ -110,12 +118,13 @@ public void testLicenseChanges() {
110118
public void testDatabaseChanges() {
111119
final XPackLicenseState licenseState = getAlwaysValidLicense();
112120
ClusterService clusterService = createClusterService(true, false);
113-
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, false, false);
121+
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, false, false, projectResolver);
114122
EnterpriseGeoIpDownloaderLicenseListener listener = new EnterpriseGeoIpDownloaderLicenseListener(
115123
client,
116124
clusterService,
117125
threadPool,
118-
licenseState
126+
licenseState,
127+
projectResolver
119128
);
120129
listener.init();
121130
listener.licenseStateChanged();
@@ -134,12 +143,13 @@ public void testMasterChanges() {
134143
// Should never start if not master node, even if all other conditions have been met
135144
final XPackLicenseState licenseState = getAlwaysValidLicense();
136145
ClusterService clusterService = createClusterService(false, false);
137-
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, false, false);
146+
TaskStartAndRemoveMockClient client = new TaskStartAndRemoveMockClient(threadPool, false, false, projectResolver);
138147
EnterpriseGeoIpDownloaderLicenseListener listener = new EnterpriseGeoIpDownloaderLicenseListener(
139148
client,
140149
clusterService,
141150
threadPool,
142-
licenseState
151+
licenseState,
152+
projectResolver
143153
);
144154
listener.init();
145155
listener.licenseStateChanged();
@@ -172,7 +182,15 @@ private ClusterState createClusterState(boolean isMasterNode, boolean hasGeoIpDa
172182
ClusterState.Builder clusterStateBuilder = ClusterState.builder(new ClusterName("name"));
173183
if (hasGeoIpDatabases) {
174184
PersistentTasksCustomMetadata tasksCustomMetadata = new PersistentTasksCustomMetadata(1L, Map.of());
175-
clusterStateBuilder.metadata(Metadata.builder().putCustom(INGEST_GEOIP_CUSTOM_METADATA_TYPE, tasksCustomMetadata).put(idxMeta));
185+
clusterStateBuilder.metadata(
186+
Metadata.builder()
187+
.put(
188+
ProjectMetadata.builder(projectResolver.getProjectId())
189+
.putCustom(INGEST_GEOIP_CUSTOM_METADATA_TYPE, tasksCustomMetadata)
190+
.put(idxMeta)
191+
.build()
192+
)
193+
);
176194
}
177195
return clusterStateBuilder.nodes(discoveryNodesBuilder).build();
178196
}
@@ -184,8 +202,13 @@ private static class TaskStartAndRemoveMockClient extends NoOpClient {
184202
private boolean taskStartCalled = false;
185203
private boolean taskRemoveCalled = false;
186204

187-
private TaskStartAndRemoveMockClient(ThreadPool threadPool, boolean expectStartTask, boolean expectRemoveTask) {
188-
super(threadPool);
205+
private TaskStartAndRemoveMockClient(
206+
ThreadPool threadPool,
207+
boolean expectStartTask,
208+
boolean expectRemoveTask,
209+
ProjectResolver projectResolver
210+
) {
211+
super(threadPool, projectResolver);
189212
this.expectStartTask = expectStartTask;
190213
this.expectRemoveTask = expectRemoveTask;
191214
}

0 commit comments

Comments
 (0)