Skip to content

Commit 3559fa0

Browse files
jtuglu1jtuglu1
committed
more changes
1 parent aa5eba1 commit 3559fa0

16 files changed

Lines changed: 299 additions & 39 deletions

File tree

‎extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/execution/KubernetesTaskExecutionConfigResource.java‎

Lines changed: 21 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -25,9 +25,11 @@
2525
import org.apache.druid.audit.AuditManager;
2626
import org.apache.druid.common.config.ConfigManager;
2727
import org.apache.druid.common.config.JacksonConfigManager;
28+
import org.apache.druid.error.DruidException;
2829
import org.apache.druid.java.util.common.Intervals;
2930
import org.apache.druid.java.util.common.logger.Logger;
3031
import org.apache.druid.server.http.DynamicConfigEtagHelper;
32+
import org.apache.druid.server.http.ServletResourceUtils;
3133
import org.apache.druid.server.http.security.ConfigResourceFilter;
3234
import org.apache.druid.server.security.AuthorizationUtils;
3335
import org.joda.time.Interval;
@@ -85,24 +87,27 @@ public Response setExecutionConfig(
8587
@Context final HttpServletRequest req
8688
)
8789
{
88-
KubernetesTaskRunnerDynamicConfig currentConfig = getDynamicConfig();
89-
KubernetesTaskRunnerDynamicConfig mergedConfig = dynamicConfig;
90+
try {
91+
final KubernetesTaskRunnerDynamicConfig currentConfig = getDynamicConfig();
92+
final KubernetesTaskRunnerDynamicConfig mergedConfig = currentConfig == null
93+
? dynamicConfig
94+
: currentConfig.merge(dynamicConfig);
95+
final ConfigManager.SetResult setResult = configManager.setIfMatch(
96+
KubernetesTaskRunnerDynamicConfig.CONFIG_KEY,
97+
DynamicConfigEtagHelper.getIfMatch(req),
98+
mergedConfig,
99+
AuthorizationUtils.buildAuditInfo(req)
100+
);
101+
if (setResult.isOk()) {
102+
log.info("Updating K8s execution configs: %s", mergedConfig);
90103

91-
if (currentConfig != null) {
92-
mergedConfig = currentConfig.merge(dynamicConfig);
104+
return Response.ok().build();
105+
} else {
106+
return DynamicConfigEtagHelper.toErrorResponse(setResult);
107+
}
93108
}
94-
final ConfigManager.SetResult setResult = configManager.setIfMatch(
95-
KubernetesTaskRunnerDynamicConfig.CONFIG_KEY,
96-
DynamicConfigEtagHelper.getIfMatch(req),
97-
mergedConfig,
98-
AuthorizationUtils.buildAuditInfo(req)
99-
);
100-
if (setResult.isOk()) {
101-
log.info("Updating K8s execution configs: %s", mergedConfig);
102-
103-
return Response.ok().build();
104-
} else {
105-
return DynamicConfigEtagHelper.toErrorResponse(setResult);
109+
catch (DruidException e) {
110+
return ServletResourceUtils.buildErrorResponseFrom(e);
106111
}
107112
}
108113

‎extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/execution/KubernetesTaskExecutionConfigResourceTest.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.druid.audit.AuditManager;
2323
import org.apache.druid.common.config.ConfigManager;
2424
import org.apache.druid.common.config.JacksonConfigManager;
25+
import org.apache.druid.error.ErrorResponse;
2526
import org.apache.druid.server.security.AuthConfig;
2627
import org.apache.druid.server.security.AuthorizationUtils;
2728
import org.easymock.EasyMock;
@@ -30,6 +31,7 @@
3031
import org.junit.jupiter.api.Test;
3132

3233
import javax.servlet.http.HttpServletRequest;
34+
import javax.ws.rs.core.HttpHeaders;
3335
import javax.ws.rs.core.Response;
3436
import java.util.concurrent.atomic.AtomicReference;
3537

@@ -107,6 +109,30 @@ public void setExecutionConfigFailedUpdate()
107109
Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), result.getStatus());
108110
}
109111

112+
@Test
113+
public void setExecutionConfigWithBlankIfMatchReturnsBadRequest()
114+
{
115+
EasyMock.expect(configManager.watch(
116+
KubernetesTaskRunnerDynamicConfig.CONFIG_KEY,
117+
KubernetesTaskRunnerDynamicConfig.class
118+
)).andReturn(new AtomicReference<>(null));
119+
final KubernetesTaskExecutionConfigResource testedResource = new KubernetesTaskExecutionConfigResource(
120+
configManager,
121+
auditManager
122+
);
123+
EasyMock.expect(req.getHeader(HttpHeaders.IF_MATCH)).andReturn(" ").once();
124+
EasyMock.replay(req, configManager, auditManager, dynamicConfig);
125+
126+
final Response result = testedResource.setExecutionConfig(dynamicConfig, req);
127+
128+
Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), result.getStatus());
129+
Assertions.assertTrue(result.getEntity() instanceof ErrorResponse);
130+
Assertions.assertEquals(
131+
"If-Match header must not be blank",
132+
((ErrorResponse) result.getEntity()).getUnderlyingException().getMessage()
133+
);
134+
}
135+
110136
@Test
111137
public void setExecutionConfig_MergeUsesCurrentCapacityWhenRequestCapacityNull()
112138
{

‎indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResource.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -123,9 +123,12 @@ public Response updateClusterCompactionConfig(
123123
)
124124
{
125125
final AuditInfo auditInfo = AuthorizationUtils.buildAuditInfo(req);
126-
final String ifMatch = DynamicConfigEtagHelper.getIfMatch(req);
127126
return ServletResourceUtils.buildUpdateResponse(
128-
() -> configManager.updateClusterCompactionConfig(updatePayload, ifMatch, auditInfo)
127+
() -> configManager.updateClusterCompactionConfig(
128+
updatePayload,
129+
DynamicConfigEtagHelper.getIfMatch(req),
130+
auditInfo
131+
)
129132
);
130133
}
131134

‎indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java‎

Lines changed: 16 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -459,18 +459,23 @@ public Response setWorkerConfig(
459459
@Context final HttpServletRequest req
460460
)
461461
{
462-
final SetResult setResult = configManager.setIfMatch(
463-
WorkerBehaviorConfig.CONFIG_KEY,
464-
DynamicConfigEtagHelper.getIfMatch(req),
465-
workerBehaviorConfig,
466-
AuthorizationUtils.buildAuditInfo(req)
467-
);
468-
if (setResult.isOk()) {
469-
log.info("Updating Worker configs: %s", workerBehaviorConfig);
462+
try {
463+
final SetResult setResult = configManager.setIfMatch(
464+
WorkerBehaviorConfig.CONFIG_KEY,
465+
DynamicConfigEtagHelper.getIfMatch(req),
466+
workerBehaviorConfig,
467+
AuthorizationUtils.buildAuditInfo(req)
468+
);
469+
if (setResult.isOk()) {
470+
log.info("Updating Worker configs: %s", workerBehaviorConfig);
470471

471-
return Response.ok().build();
472-
} else {
473-
return DynamicConfigEtagHelper.toErrorResponse(setResult);
472+
return Response.ok().build();
473+
} else {
474+
return DynamicConfigEtagHelper.toErrorResponse(setResult);
475+
}
476+
}
477+
catch (DruidException e) {
478+
return ServletResourceUtils.buildErrorResponseFrom(e);
474479
}
475480
}
476481

‎indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@
5555
import org.junit.Test;
5656

5757
import javax.servlet.http.HttpServletRequest;
58+
import javax.ws.rs.core.HttpHeaders;
5859
import javax.ws.rs.core.Response;
5960
import java.util.List;
6061
import java.util.Map;
@@ -171,6 +172,21 @@ public void test_updateClusterConfig()
171172
Assert.assertEquals(Map.of("success", true), response.getEntity());
172173
}
173174

175+
@Test
176+
public void test_updateClusterConfigWithBlankIfMatchReturnsBadRequest()
177+
{
178+
setupMockRequestForAudit();
179+
EasyMock.expect(httpRequest.getHeader(HttpHeaders.IF_MATCH)).andReturn(" ").once();
180+
replayAll();
181+
182+
final Response response = compactionResource.updateClusterCompactionConfig(
183+
new ClusterCompactionConfig(0.5, 10, null, true, CompactionEngine.MSQ, true),
184+
httpRequest
185+
);
186+
187+
verifyInvalidInputResponse(response, "If-Match header must not be blank");
188+
}
189+
174190
@Test
175191
public void test_getClusterConfig()
176192
{

‎indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.druid.audit.AuditManager;
3030
import org.apache.druid.common.config.JacksonConfigManager;
3131
import org.apache.druid.error.DruidException;
32+
import org.apache.druid.error.ErrorResponse;
3233
import org.apache.druid.error.InvalidInput;
3334
import org.apache.druid.indexer.RunnerTaskState;
3435
import org.apache.druid.indexer.TaskInfo;
@@ -90,6 +91,7 @@
9091

9192
import javax.servlet.http.HttpServletRequest;
9293
import javax.ws.rs.WebApplicationException;
94+
import javax.ws.rs.core.HttpHeaders;
9395
import javax.ws.rs.core.Response;
9496
import javax.ws.rs.core.Response.Status;
9597
import java.util.Arrays;
@@ -1363,6 +1365,23 @@ public void testGetTotalWorkerCapacityWithMaximumCapacity()
13631365
Assert.assertEquals(expectedWorkerCapacityWithAutoscale, ((TotalWorkerCapacityResponse) response.getEntity()).getMaximumCapacityWithAutoScale());
13641366
}
13651367

1368+
@Test
1369+
public void testSetWorkerConfigWithBlankIfMatchReturnsBadRequest()
1370+
{
1371+
final WorkerBehaviorConfig workerBehaviorConfig = EasyMock.createMock(WorkerBehaviorConfig.class);
1372+
EasyMock.expect(req.getHeader(HttpHeaders.IF_MATCH)).andReturn(" ").once();
1373+
replayAll();
1374+
1375+
final Response response = overlordResource.setWorkerConfig(workerBehaviorConfig, req);
1376+
1377+
Assert.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus());
1378+
Assert.assertTrue(response.getEntity() instanceof ErrorResponse);
1379+
Assert.assertEquals(
1380+
"If-Match header must not be blank",
1381+
((ErrorResponse) response.getEntity()).getUnderlyingException().getMessage()
1382+
);
1383+
}
1384+
13661385
@Test
13671386
public void testResourceActionsForTaskWithInputTypeAndInputSecurityEnabled()
13681387
{

‎processing/src/main/java/org/apache/druid/common/config/ConfigManager.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -200,6 +200,11 @@ public byte[] getCurrentBytes(final String key)
200200
return dbConnector.lookup(configTable, "name", "payload", key);
201201
}
202202

203+
public boolean isCompareAndSwapEnabled()
204+
{
205+
return config.get().isEnableCompareAndSwap();
206+
}
207+
203208
public <T> SetResult set(final String key, final ConfigSerde<T> serde, @Nullable final byte[] oldValue, final T newObject)
204209
{
205210
if (newObject == null || !started) {

‎processing/src/main/java/org/apache/druid/common/config/JacksonConfigManager.java‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -133,17 +133,22 @@ public byte[] getCurrentBytes(String key)
133133
return configManager.getCurrentBytes(key);
134134
}
135135

136+
public boolean isCompareAndSwapEnabled()
137+
{
138+
return configManager.isCompareAndSwapEnabled();
139+
}
140+
136141
/**
137142
* Set the config, optionally guarded by an {@code If-Match}-style
138143
* precondition. When {@code ifMatchEtag} is {@code null}, behaves like
139144
* {@link #set(String, Object, AuditInfo)}. Otherwise the write only commits
140145
* if the currently stored bytes hash to {@code ifMatchEtag}; on mismatch the
141146
* result reports {@link SetResult#isPreconditionFailed() preconditionFailed}.
142147
*
143-
* <p>The precondition is enforced via metadata-store CAS, so it only holds
144-
* when {@code druid.manager.config.enableCompareAndSwap} is true (the
145-
* default). With CAS disabled, the {@code If-Match} check degrades to a
146-
* best-effort TOCTOU read and concurrent writers may still race past it.
148+
* <p>The precondition is enforced via metadata-store CAS, so conditional
149+
* writes require {@code druid.manager.config.enableCompareAndSwap} to be
150+
* true (the default). With CAS disabled, {@code If-Match} writes fail as a
151+
* precondition failure instead of silently degrading to last-writer-wins.
147152
*/
148153
public <T> SetResult setIfMatch(
149154
String key,
@@ -158,6 +163,13 @@ public <T> SetResult setIfMatch(
158163
if (ifMatchEtag == null) {
159164
return set(key, newValue, auditInfo);
160165
}
166+
if (!configManager.isCompareAndSwapEnabled()) {
167+
return SetResult.preconditionFailed(
168+
new IllegalStateException(
169+
"If-Match requires druid.manager.config.enableCompareAndSwap to be enabled for key[" + key + "]"
170+
)
171+
);
172+
}
161173
final byte[] currentBytes = configManager.getCurrentBytes(key);
162174
if (!ConfigEtag.matches(ifMatchEtag, currentBytes)) {
163175
return SetResult.preconditionFailed(

‎processing/src/test/java/org/apache/druid/common/config/JacksonConfigManagerTest.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,7 @@ public void testSetIfMatchPreconditionPassesForMatchingEtag()
108108
TestConfig val = new TestConfig("v", "s", 1);
109109
AuditInfo auditInfo = new AuditInfo("a", "i", "c", "ip");
110110
byte[] currentBytes = "current".getBytes(java.nio.charset.StandardCharsets.UTF_8);
111+
Mockito.when(mockConfigManager.isCompareAndSwapEnabled()).thenReturn(true);
111112
Mockito.when(mockConfigManager.getCurrentBytes(key)).thenReturn(currentBytes);
112113
Mockito.when(mockConfigManager.set(
113114
Mockito.eq(key),
@@ -135,6 +136,7 @@ public void testSetIfMatchPreconditionFailsForMismatchedEtag()
135136
String key = "key";
136137
TestConfig val = new TestConfig("v", "s", 1);
137138
AuditInfo auditInfo = new AuditInfo("a", "i", "c", "ip");
139+
Mockito.when(mockConfigManager.isCompareAndSwapEnabled()).thenReturn(true);
138140
Mockito.when(mockConfigManager.getCurrentBytes(key)).thenReturn("current".getBytes(java.nio.charset.StandardCharsets.UTF_8));
139141

140142
ConfigManager.SetResult result = jacksonConfigManager.setIfMatch(
@@ -154,6 +156,28 @@ public void testSetIfMatchPreconditionFailsForMismatchedEtag()
154156
);
155157
}
156158

159+
@Test
160+
public void testSetIfMatchPreconditionFailsWhenCompareAndSwapDisabled()
161+
{
162+
final String key = "key";
163+
final TestConfig val = new TestConfig("v", "s", 1);
164+
final AuditInfo auditInfo = new AuditInfo("a", "i", "c", "ip");
165+
Mockito.when(mockConfigManager.isCompareAndSwapEnabled()).thenReturn(false);
166+
167+
final ConfigManager.SetResult result = jacksonConfigManager.setIfMatch(key, "\"etag\"", val, auditInfo);
168+
169+
Assertions.assertFalse(result.isOk());
170+
Assertions.assertTrue(result.isPreconditionFailed());
171+
Mockito.verify(mockConfigManager, Mockito.never()).getCurrentBytes(Mockito.anyString());
172+
Mockito.verify(mockConfigManager, Mockito.never()).set(
173+
Mockito.anyString(),
174+
Mockito.any(ConfigSerde.class),
175+
Mockito.any(byte[].class),
176+
Mockito.any()
177+
);
178+
Mockito.verify(mockAuditManager, Mockito.never()).doAudit(Mockito.any(AuditEntry.class));
179+
}
180+
157181
@Test
158182
public void testConvertByteToConfigWithNullConfigInByte()
159183
{

‎server/src/main/java/org/apache/druid/server/coordinator/CoordinatorConfigManager.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,15 @@ public ConfigManager.SetResult getAndUpdateCompactionConfig(
152152
AuditInfo auditInfo
153153
)
154154
{
155+
if (ifMatchEtag != null && !jacksonConfigManager.isCompareAndSwapEnabled()) {
156+
return ConfigManager.SetResult.preconditionFailed(
157+
new IllegalStateException(
158+
"If-Match requires druid.manager.config.enableCompareAndSwap to be enabled for key["
159+
+ DruidCompactionConfig.CONFIG_KEY
160+
+ "]"
161+
)
162+
);
163+
}
155164
// Fetch the bytes and use to build the current config and perform compare-and-swap.
156165
// This avoids failures in ConfigManager while updating configs previously
157166
// persisted by older versions of Druid which didn't have fields such as 'granularitySpec'.

0 commit comments

Comments
 (0)