Skip to content

Commit 47c1114

Browse files
samxbrsamxbr
committed
Make EnterpriseGeoIpDownloaderLicenseListener project aware
1 parent 41a47c2 commit 47c1114

File tree

3 files changed

+92
-45
lines changed

3 files changed

+92
-45
lines changed

‎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: 56 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,12 @@
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.ProjectState;
21+
import org.elasticsearch.cluster.metadata.ProjectId;
22+
import org.elasticsearch.cluster.project.ProjectResolver;
2123
import org.elasticsearch.cluster.service.ClusterService;
24+
import org.elasticsearch.core.FixForMultiProject;
2225
import org.elasticsearch.ingest.EnterpriseGeoIpTask.EnterpriseGeoIpTaskParams;
2326
import org.elasticsearch.license.License;
2427
import org.elasticsearch.license.LicenseStateListener;
@@ -31,12 +34,14 @@
3134
import org.elasticsearch.xpack.core.XPackField;
3235

3336
import java.util.Objects;
37+
import java.util.concurrent.ConcurrentHashMap;
38+
import java.util.concurrent.ConcurrentMap;
3439

3540
import static org.elasticsearch.ingest.EnterpriseGeoIpTask.ENTERPRISE_GEOIP_DOWNLOADER;
3641

3742
public class EnterpriseGeoIpDownloaderLicenseListener implements LicenseStateListener, ClusterStateListener {
3843
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
44+
// Note: This custom type is IngestGeoIpMetadata.TYPE, but that class is not exposed to this plugin
4045
static final String INGEST_GEOIP_CUSTOM_METADATA_TYPE = "ingest_geoip";
4146

4247
private final PersistentTasksService persistentTasksService;
@@ -47,18 +52,21 @@ public class EnterpriseGeoIpDownloaderLicenseListener implements LicenseStateLis
4752
XPackField.ENTERPRISE_GEOIP_DOWNLOADER,
4853
License.OperationMode.PLATINUM
4954
);
50-
private volatile boolean licenseIsValid = false;
51-
private volatile boolean hasIngestGeoIpMetadata = false;
55+
private final ConcurrentMap<ProjectId, Boolean> licenseIsValid = new ConcurrentHashMap<>();
56+
private final ConcurrentMap<ProjectId, Boolean> hasIngestGeoIpMetadata = new ConcurrentHashMap<>();
57+
private final ProjectResolver projectResolver;
5258

5359
protected EnterpriseGeoIpDownloaderLicenseListener(
5460
Client client,
5561
ClusterService clusterService,
5662
ThreadPool threadPool,
57-
XPackLicenseState licenseState
63+
XPackLicenseState licenseState,
64+
ProjectResolver projectResolver
5865
) {
5966
this.persistentTasksService = new PersistentTasksService(clusterService, threadPool, client);
6067
this.clusterService = clusterService;
6168
this.licenseState = licenseState;
69+
this.projectResolver = projectResolver;
6270
}
6371

6472
private volatile boolean licenseStateListenerRegistered;
@@ -74,47 +82,54 @@ void listenForLicenseStateChanges() {
7482
licenseState.addListener(this);
7583
}
7684

85+
@FixForMultiProject(description = "Replace DEFAULT project after license is project-aware")
7786
@Override
7887
public void licenseStateChanged() {
79-
licenseIsValid = ENTERPRISE_GEOIP_FEATURE.checkWithoutTracking(licenseState);
80-
maybeUpdateTaskState(clusterService.state());
88+
licenseIsValid.put(ProjectId.DEFAULT, ENTERPRISE_GEOIP_FEATURE.checkWithoutTracking(licenseState));
89+
maybeUpdateTaskState(clusterService.state().projectState(ProjectId.DEFAULT));
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);
88-
final boolean masterNodeChanged = Objects.equals(
89-
event.state().nodes().getMasterNode(),
90-
event.previousState().nodes().getMasterNode()
91-
) == 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-
}
94+
event.state().forEachProject(projectState -> {
95+
ProjectId projectId = projectState.projectId();
96+
final boolean hasMetadata = event.state().metadata().getProject(projectId).custom(INGEST_GEOIP_CUSTOM_METADATA_TYPE) != null;
97+
hasIngestGeoIpMetadata.put(projectId, hasMetadata);
98+
final boolean ingestGeoIpCustomMetaChangedInEvent = event.metadataChanged()
99+
&& event.customMetadataChanged(projectId, INGEST_GEOIP_CUSTOM_METADATA_TYPE);
100+
final boolean masterNodeChanged = Objects.equals(
101+
event.state().nodes().getMasterNode(),
102+
event.previousState().nodes().getMasterNode()
103+
) == false;
104+
/*
105+
* We don't want to potentially start the task on every cluster state change, so only maybeUpdateTaskState
106+
* if this cluster change event involved the modification of custom geoip metadata OR a master node change
107+
*/
108+
if (ingestGeoIpCustomMetaChangedInEvent || (masterNodeChanged && hasIngestGeoIpMetadata.getOrDefault(projectId, false))) {
109+
maybeUpdateTaskState(projectState);
110+
}
111+
});
99112
}
100113

101-
private void maybeUpdateTaskState(ClusterState state) {
114+
private void maybeUpdateTaskState(ProjectState projectState) {
115+
ProjectId projectId = projectState.projectId();
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 (projectState.cluster().nodes().isLocalNodeElectedMaster()) {
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,14 @@ private void ensureTaskStopped() {
137152
}
138153
}
139154
);
140-
persistentTasksService.sendRemoveRequest(ENTERPRISE_GEOIP_DOWNLOADER, MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT, listener);
155+
persistentTasksService.sendRemoveRequest(
156+
getTaskId(projectId, projectResolver.supportsMultipleProjects()),
157+
MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
158+
listener
159+
);
160+
}
161+
162+
protected static String getTaskId(ProjectId projectId, boolean supportsMultipleProjects) {
163+
return supportsMultipleProjects ? projectId + "/" + ENTERPRISE_GEOIP_DOWNLOADER : ENTERPRISE_GEOIP_DOWNLOADER;
141164
}
142165
}

‎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.FixForMultiProject;
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+
@FixForMultiProject
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)