Skip to content

Commit 3b29c6d

Browse files
luoluoyuyuluoluoyuyu
committed
Prevent pipe sink task id from logging secrets (#17981)
* Prevent pipe sink task id from logging secrets * Fix pipe sink subtask compatibility
1 parent ecdc547 commit 3b29c6d

14 files changed

Lines changed: 180 additions & 51 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeTypeConversionISessionIT.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@
5656
import java.util.Date;
5757
import java.util.List;
5858
import java.util.Random;
59+
import java.util.UUID;
5960
import java.util.concurrent.TimeUnit;
6061

6162
import static org.awaitility.Awaitility.await;
@@ -324,7 +325,7 @@ private void prepareTypeConversionTest(
324325
generateMeasurementSchemas();
325326

326327
// Generate createTimeSeries in sender and receiver
327-
String uuid = "bcdedit";
328+
String uuid = "bcdedit" + UUID.randomUUID().toString().replace("-", "");
328329
for (Pair<MeasurementSchema, MeasurementSchema> pair : measurementSchemas) {
329330
createTimeSeries(uuid, pair.left.getMeasurementName(), pair.left.getType().name(), senderEnv);
330331
createTimeSeries(

‎iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import java.util.HashMap;
3030
import java.util.HashSet;
3131
import java.util.List;
32+
import java.util.Locale;
3233
import java.util.Map;
3334
import java.util.Map.Entry;
3435
import java.util.Objects;
@@ -433,13 +434,14 @@ public static class ValueHider {
433434
}
434435

435436
static String hide(final String key, final String value) {
436-
if (Objects.isNull(key)) {
437-
return value;
438-
}
439-
if (KEYS.contains(KeyReducer.reduce(key))) {
437+
if (isHiddenKey(key)) {
440438
return PLACEHOLDER;
441439
}
442440
return value;
443441
}
442+
443+
public static boolean isHiddenKey(final String key) {
444+
return Objects.nonNull(key) && KEYS.contains(KeyReducer.reduce(key).toLowerCase(Locale.ROOT));
445+
}
444446
}
445447
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -260,7 +260,7 @@ protected boolean executeOnce() throws Exception {
260260
throw new PipeException(
261261
String.format(
262262
"Exception in pipe process, subtask: %s, last event: %s, root cause: %s",
263-
taskID,
263+
getDisplayTaskID(),
264264
lastEvent instanceof EnrichedEvent
265265
? ((EnrichedEvent) lastEvent).coreReportMessage()
266266
: lastEvent,
@@ -300,7 +300,7 @@ public void close() {
300300
} catch (final Exception e) {
301301
LOGGER.info(
302302
DataNodePipeMessages.EXCEPTION_OCCURRED_WHEN_CLOSING_PIPE_PROCESSOR_SUBTASK,
303-
taskID,
303+
getDisplayTaskID(),
304304
ErrorHandlingCommonUtils.getRootCause(e).getMessage(),
305305
e);
306306
} finally {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
6060

6161
// Record these variables to provide corresponding value to tag key of monitoring metrics
6262
private final String attributeSortedString;
63+
private final String attributeDisplayString;
6364
private final int sinkIndex;
6465

6566
// Now parallel connectors run the same time, thus the heartbeat events are not sure
@@ -75,8 +76,27 @@ public PipeSinkSubtask(
7576
final int sinkIndex,
7677
final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
7778
final PipeConnector outputPipeConnector) {
79+
this(
80+
taskID,
81+
creationTime,
82+
attributeSortedString,
83+
attributeSortedString,
84+
sinkIndex,
85+
inputPendingQueue,
86+
outputPipeConnector);
87+
}
88+
89+
public PipeSinkSubtask(
90+
final String taskID,
91+
final long creationTime,
92+
final String attributeSortedString,
93+
final String attributeDisplayString,
94+
final int sinkIndex,
95+
final UnboundedBlockingPendingQueue<Event> inputPendingQueue,
96+
final PipeConnector outputPipeConnector) {
7897
super(taskID, creationTime, outputPipeConnector);
7998
this.attributeSortedString = attributeSortedString;
99+
this.attributeDisplayString = attributeDisplayString;
80100
this.sinkIndex = sinkIndex;
81101
this.inputPendingQueue = inputPendingQueue;
82102

@@ -156,7 +176,7 @@ private void transferHeartbeatEvent(final PipeHeartbeatEvent event) {
156176
DataNodePipeMessages.PIPECONNECTOR
157177
+ outputPipeSink.getClass().getName()
158178
+ "(id: "
159-
+ taskID
179+
+ getDisplayTaskID()
160180
+ ")"
161181
+ " heartbeat failed, or encountered failure when transferring generic event. Failure: "
162182
+ e.getMessage(),
@@ -181,13 +201,13 @@ public void close() {
181201
outputPipeSink.close();
182202
LOGGER.info(
183203
DataNodePipeMessages.PIPE_CONNECTOR_SUBTASK_WAS_CLOSED_WITHIN_MS,
184-
taskID,
204+
getDisplayTaskID(),
185205
outputPipeSink,
186206
System.currentTimeMillis() - startTime);
187207
} catch (final Exception e) {
188208
LOGGER.info(
189209
DataNodePipeMessages.EXCEPTION_OCCURRED_WHEN_CLOSING_PIPE_CONNECTOR_SUBTASK,
190-
taskID,
210+
getDisplayTaskID(),
191211
ErrorHandlingCommonUtils.getRootCause(e).getMessage(),
192212
e);
193213
} finally {
@@ -366,4 +386,14 @@ protected void report(final EnrichedEvent event, final PipeRuntimeException exce
366386
lastExceptionTime = Long.MAX_VALUE;
367387
PipeDataNodeAgent.runtime().report(event, exception);
368388
}
389+
390+
@Override
391+
public String getDisplayTaskID() {
392+
return generateDisplayTaskID(attributeDisplayString, creationTime, sinkIndex);
393+
}
394+
395+
static String generateDisplayTaskID(
396+
final String attributeDisplayString, final long creationTime, final int sinkIndex) {
397+
return String.format("%s_%s_%s", attributeDisplayString, creationTime, sinkIndex);
398+
}
369399
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ public synchronized void register() {
7575
registeredTaskCount++;
7676
LOGGER.info(
7777
DataNodePipeMessages.REGISTER_SUBTASK_RUNNINGTASKCOUNT_REGISTEREDTASKCOUNT,
78-
subtask,
78+
subtask.getDisplayTaskID(),
7979
runningTaskCount,
8080
registeredTaskCount);
8181
}
@@ -112,7 +112,7 @@ public synchronized boolean deregister(final CommitterKey committerKey) {
112112
registeredTaskCount--;
113113
LOGGER.info(
114114
DataNodePipeMessages.DEREGISTER_SUBTASK_RUNNINGTASKCOUNT_REGISTEREDTASKCOUNT,
115-
subtask,
115+
subtask.getDisplayTaskID(),
116116
runningTaskCount,
117117
registeredTaskCount);
118118
}
@@ -135,7 +135,7 @@ public synchronized void start() {
135135
runningTaskCount++;
136136
LOGGER.info(
137137
DataNodePipeMessages.START_SUBTASK_RUNNINGTASKCOUNT_REGISTEREDTASKCOUNT,
138-
subtask,
138+
subtask.getDisplayTaskID(),
139139
runningTaskCount,
140140
registeredTaskCount);
141141
}
@@ -152,7 +152,7 @@ public synchronized void stop() {
152152
runningTaskCount--;
153153
LOGGER.info(
154154
DataNodePipeMessages.STOP_SUBTASK_RUNNINGTASKCOUNT_REGISTEREDTASKCOUNT,
155-
subtask,
155+
subtask.getDisplayTaskID(),
156156
runningTaskCount,
157157
registeredTaskCount);
158158
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java‎

Lines changed: 46 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,8 @@ public class PipeSinkSubtaskManager {
6464
private final Map<String, List<PipeSinkSubtaskLifeCycle>>
6565
attributeSortedString2SubtaskLifeCycleMap = new HashMap<>();
6666

67+
private final Map<String, String> attributeSortedString2DisplayString = new HashMap<>();
68+
6769
public synchronized String register(
6870
final Supplier<? extends PipeSinkSubtaskExecutor> executorSupplier,
6971
final PipeParameters pipeSinkParameters,
@@ -92,19 +94,14 @@ public synchronized String register(
9294
final int sinkNum;
9395
boolean realTimeFirst = false;
9496
String attributeSortedString = generateAttributeSortedString(pipeSinkParameters);
97+
final String attributeDisplayString = generateAttributeDisplayString(pipeSinkParameters);
9598
if (isDataRegionSink) {
9699
sinkNum =
97100
pipeSinkParameters.getIntOrDefault(
98101
Arrays.asList(
99102
PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY,
100103
PipeSinkConstant.SINK_IOTDB_PARALLEL_TASKS_KEY),
101-
PipeSinkConstant.SINGLE_THREAD_DEFAULT_SINK.contains(
102-
pipeSinkParameters
103-
.getStringOrDefault(
104-
Arrays.asList(
105-
PipeSinkConstant.CONNECTOR_KEY, PipeSinkConstant.SINK_KEY),
106-
BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName())
107-
.toLowerCase())
104+
PipeSinkConstant.SINGLE_THREAD_DEFAULT_SINK.contains(connectorKey)
108105
? 1
109106
: PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);
110107
realTimeFirst =
@@ -120,7 +117,9 @@ public synchronized String register(
120117
sinkNum = 1;
121118
attributeSortedString = "schema_" + attributeSortedString;
122119
}
123-
environment.setAttributeSortedString(attributeSortedString);
120+
final String attributeDisplayStringWithPrefix =
121+
isDataRegionSink ? "data_" + attributeDisplayString : "schema_" + attributeDisplayString;
122+
environment.setAttributeSortedString(attributeDisplayStringWithPrefix);
124123

125124
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
126125
final PipeSinkSubtaskExecutor executor = executorSupplier.get();
@@ -138,6 +137,11 @@ public synchronized String register(
138137
}
139138

140139
for (int sinkIndex = 0; sinkIndex < sinkNum; sinkIndex++) {
140+
final String taskID =
141+
String.format(
142+
"%s_%s_%s",
143+
attributeDisplayStringWithPrefix, environment.getCreationTime(), sinkIndex);
144+
141145
final PipeConnector pipeSink =
142146
isDataRegionSink
143147
? PipeDataNodeAgent.plugin().dataRegion().reflectSink(pipeSinkParameters)
@@ -168,10 +172,10 @@ public synchronized String register(
168172
// 2. Construct PipeConnectorSubtaskLifeCycle to manage PipeConnectorSubtask's life cycle
169173
final PipeSinkSubtask pipeSinkSubtask =
170174
new PipeSinkSubtask(
171-
String.format(
172-
"%s_%s_%s", attributeSortedString, environment.getCreationTime(), sinkIndex),
175+
taskID,
173176
environment.getCreationTime(),
174177
attributeSortedString,
178+
attributeDisplayStringWithPrefix,
175179
sinkIndex,
176180
pendingQueue,
177181
pipeSink);
@@ -182,11 +186,13 @@ public synchronized String register(
182186

183187
LOGGER.info(
184188
DataNodePipeMessages.PIPE_SINK_SUBTASKS_WITH_ATTRIBUTES_IS_BOUNDED,
185-
attributeSortedString,
189+
attributeDisplayStringWithPrefix,
186190
executor.getWorkingThreadName(),
187191
executor.getCallbackThreadName());
188192
attributeSortedString2SubtaskLifeCycleMap.put(
189193
attributeSortedString, pipeSinkSubtaskLifeCycleList);
194+
attributeSortedString2DisplayString.put(
195+
attributeSortedString, attributeDisplayStringWithPrefix);
190196
}
191197

192198
for (final PipeSinkSubtaskLifeCycle lifeCycle :
@@ -203,7 +209,7 @@ public synchronized void deregister(
203209
final int regionId,
204210
final String attributeSortedString) {
205211
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
206-
throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE + attributeSortedString);
212+
throwNoSuchSubtaskException(attributeSortedString);
207213
}
208214

209215
final List<PipeSinkSubtaskLifeCycle> lifeCycles =
@@ -219,6 +225,7 @@ public synchronized void deregister(
219225

220226
if (lifeCycles.isEmpty()) {
221227
attributeSortedString2SubtaskLifeCycleMap.remove(attributeSortedString);
228+
attributeSortedString2DisplayString.remove(attributeSortedString);
222229
executor.shutdown();
223230
LOGGER.info(
224231
DataNodePipeMessages.THE_EXECUTOR_AND_HAS_BEEN_SUCCESSFULLY_SHUTDOWN,
@@ -234,7 +241,7 @@ public synchronized void deregister(
234241

235242
public synchronized void start(final String attributeSortedString) {
236243
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
237-
throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE + attributeSortedString);
244+
throwNoSuchSubtaskException(attributeSortedString);
238245
}
239246

240247
for (final PipeSinkSubtaskLifeCycle lifeCycle :
@@ -245,7 +252,7 @@ public synchronized void start(final String attributeSortedString) {
245252

246253
public synchronized void stop(final String attributeSortedString) {
247254
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
248-
throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE + attributeSortedString);
255+
throwNoSuchSubtaskException(attributeSortedString);
249256
}
250257

251258
for (final PipeSinkSubtaskLifeCycle lifeCycle :
@@ -258,7 +265,8 @@ public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
258265
final String attributeSortedString) {
259266
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
260267
throw new PipeException(
261-
DataNodePipeMessages.FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK + attributeSortedString);
268+
DataNodePipeMessages.FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK
269+
+ getDisplayStringForException(attributeSortedString));
262270
}
263271

264272
// All subtasks share the same pending queue
@@ -268,13 +276,35 @@ public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
268276
.getPendingQueue();
269277
}
270278

271-
private String generateAttributeSortedString(final PipeParameters pipeConnectorParameters) {
279+
private static String generateAttributeSortedString(
280+
final PipeParameters pipeConnectorParameters) {
272281
final TreeMap<String, String> sortedStringSourceMap =
273282
new TreeMap<>(pipeConnectorParameters.getAttribute());
274283
sortedStringSourceMap.remove(SystemConstant.RESTART_OR_NEWLY_ADDED_KEY);
275284
return sortedStringSourceMap.toString();
276285
}
277286

287+
/**
288+
* Attribute string for logs, metrics and exception messages with sensitive attributes removed.
289+
*/
290+
static String generateAttributeDisplayString(final PipeParameters pipeConnectorParameters) {
291+
final TreeMap<String, String> filteredAttributes =
292+
new TreeMap<>(pipeConnectorParameters.getAttribute());
293+
filteredAttributes.remove(SystemConstant.RESTART_OR_NEWLY_ADDED_KEY);
294+
filteredAttributes.keySet().removeIf(PipeParameters.ValueHider::isHiddenKey);
295+
return filteredAttributes.toString();
296+
}
297+
298+
private void throwNoSuchSubtaskException(final String attributeSortedString) {
299+
throw new PipeException(
300+
FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE
301+
+ getDisplayStringForException(attributeSortedString));
302+
}
303+
304+
private String getDisplayStringForException(final String attributeSortedString) {
305+
return attributeSortedString2DisplayString.getOrDefault(attributeSortedString, "unknown");
306+
}
307+
278308
///////////////////////// Singleton Instance Holder /////////////////////////
279309

280310
private PipeSinkSubtaskManager() {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,9 @@ public void register(final PipeSinkSubtask pipeSinkSubtask) {
110110

111111
public void deregister(final String taskID) {
112112
if (!connectorMap.containsKey(taskID)) {
113-
LOGGER.warn(DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_SCHEMA_REGION_CONNECTOR, taskID);
113+
LOGGER.warn(
114+
DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_SCHEMA_REGION_CONNECTOR,
115+
getDisplayTaskID(taskID));
114116
return;
115117
}
116118
if (Objects.nonNull(metricService)) {
@@ -125,12 +127,18 @@ public void markSchemaEvent(final String taskID) {
125127
}
126128
final Rate rate = schemaRateMap.get(taskID);
127129
if (rate == null) {
128-
LOGGER.info(DataNodePipeMessages.FAILED_TO_MARK_PIPE_SCHEMA_REGION_WRITE, taskID);
130+
LOGGER.info(
131+
DataNodePipeMessages.FAILED_TO_MARK_PIPE_SCHEMA_REGION_WRITE, getDisplayTaskID(taskID));
129132
return;
130133
}
131134
rate.mark();
132135
}
133136

137+
private String getDisplayTaskID(final String taskID) {
138+
final PipeSinkSubtask connector = connectorMap.get(taskID);
139+
return Objects.nonNull(connector) ? connector.getDisplayTaskID() : "unknown";
140+
}
141+
134142
//////////////////////////// singleton ////////////////////////////
135143

136144
private static class PipeSchemaRegionSinkMetricsHolder {

0 commit comments

Comments
 (0)