Skip to content

Commit 2ba2914

Browse files
luoluoyuyucursoragent
luoluoyuyu
and
cursoragent
authored
Pipe: Mask sensitive attributes in sink subtask display strings (#17737)
* Pipe: Mask sensitive attributes in sink subtask display strings Use masked PipeParameters display string for logs, metrics and subtask names while keeping unmasked sorted string for internal lifecycle map keys. Also treat scp.password as a sensitive parameter. Co-authored-by: Cursor <cursoragent@cursor.com> * Pipe: Fix sink compression timer keying and masked error paths Key compressionTimerMap by per-subtask taskID instead of masked attribute string to avoid collisions when only sensitive fields differ. Use masked display strings in subtask-not-found exceptions and pass sinkTaskId from runtime environment to IoTDB sinks for timer lookup. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent 4e3bb11 commit 2ba2914

8 files changed

Lines changed: 70 additions & 26 deletions

File tree

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -429,6 +429,7 @@ public static class ValueHider {
429429

430430
static {
431431
KEYS.add("ssl.trust-store-pwd");
432+
KEYS.add("scp.password");
432433
KEYS.add("password");
433434
}
434435

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

Lines changed: 44 additions & 10 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,6 +94,7 @@ 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(
@@ -120,7 +123,9 @@ public synchronized String register(
120123
sinkNum = 1;
121124
attributeSortedString = "schema_" + attributeSortedString;
122125
}
123-
environment.setAttributeSortedString(attributeSortedString);
126+
final String attributeDisplayStringWithPrefix =
127+
isDataRegionSink ? "data_" + attributeDisplayString : "schema_" + attributeDisplayString;
128+
environment.setAttributeSortedString(attributeDisplayStringWithPrefix);
124129

125130
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
126131
final PipeSinkSubtaskExecutor executor = executorSupplier.get();
@@ -138,6 +143,12 @@ public synchronized String register(
138143
}
139144

140145
for (int sinkIndex = 0; sinkIndex < sinkNum; sinkIndex++) {
146+
final String taskID =
147+
String.format(
148+
"%s_%s_%s",
149+
attributeDisplayStringWithPrefix, environment.getCreationTime(), sinkIndex);
150+
environment.setSinkTaskId(taskID);
151+
141152
final PipeConnector pipeSink =
142153
isDataRegionSink
143154
? PipeDataNodeAgent.plugin().dataRegion().reflectSink(pipeSinkParameters)
@@ -168,10 +179,9 @@ public synchronized String register(
168179
// 2. Construct PipeConnectorSubtaskLifeCycle to manage PipeConnectorSubtask's life cycle
169180
final PipeSinkSubtask pipeSinkSubtask =
170181
new PipeSinkSubtask(
171-
String.format(
172-
"%s_%s_%s", attributeSortedString, environment.getCreationTime(), sinkIndex),
182+
taskID,
173183
environment.getCreationTime(),
174-
attributeSortedString,
184+
attributeDisplayStringWithPrefix,
175185
sinkIndex,
176186
pendingQueue,
177187
pipeSink);
@@ -182,11 +192,13 @@ public synchronized String register(
182192

183193
LOGGER.info(
184194
DataNodePipeMessages.PIPE_SINK_SUBTASKS_WITH_ATTRIBUTES_IS_BOUNDED,
185-
attributeSortedString,
195+
attributeDisplayStringWithPrefix,
186196
executor.getWorkingThreadName(),
187197
executor.getCallbackThreadName());
188198
attributeSortedString2SubtaskLifeCycleMap.put(
189199
attributeSortedString, pipeSinkSubtaskLifeCycleList);
200+
attributeSortedString2DisplayString.put(
201+
attributeSortedString, attributeDisplayStringWithPrefix);
190202
}
191203

192204
for (final PipeSinkSubtaskLifeCycle lifeCycle :
@@ -203,7 +215,7 @@ public synchronized void deregister(
203215
final int regionId,
204216
final String attributeSortedString) {
205217
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
206-
throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE + attributeSortedString);
218+
throwNoSuchSubtaskException(attributeSortedString);
207219
}
208220

209221
final List<PipeSinkSubtaskLifeCycle> lifeCycles =
@@ -219,6 +231,7 @@ public synchronized void deregister(
219231

220232
if (lifeCycles.isEmpty()) {
221233
attributeSortedString2SubtaskLifeCycleMap.remove(attributeSortedString);
234+
attributeSortedString2DisplayString.remove(attributeSortedString);
222235
executor.shutdown();
223236
LOGGER.info(
224237
DataNodePipeMessages.THE_EXECUTOR_AND_HAS_BEEN_SUCCESSFULLY_SHUTDOWN,
@@ -234,7 +247,7 @@ public synchronized void deregister(
234247

235248
public synchronized void start(final String attributeSortedString) {
236249
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
237-
throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE + attributeSortedString);
250+
throwNoSuchSubtaskException(attributeSortedString);
238251
}
239252

240253
for (final PipeSinkSubtaskLifeCycle lifeCycle :
@@ -245,7 +258,7 @@ public synchronized void start(final String attributeSortedString) {
245258

246259
public synchronized void stop(final String attributeSortedString) {
247260
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
248-
throw new PipeException(FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE + attributeSortedString);
261+
throwNoSuchSubtaskException(attributeSortedString);
249262
}
250263

251264
for (final PipeSinkSubtaskLifeCycle lifeCycle :
@@ -258,7 +271,8 @@ public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
258271
final String attributeSortedString) {
259272
if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
260273
throw new PipeException(
261-
DataNodePipeMessages.FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK + attributeSortedString);
274+
DataNodePipeMessages.FAILED_TO_GET_PENDINGQUEUE_NO_SUCH_SUBTASK
275+
+ getDisplayStringForException(attributeSortedString));
262276
}
263277

264278
// All subtasks share the same pending queue
@@ -268,13 +282,33 @@ public UnboundedBlockingPendingQueue<Event> getPipeSinkPendingQueue(
268282
.getPendingQueue();
269283
}
270284

271-
private String generateAttributeSortedString(final PipeParameters pipeConnectorParameters) {
285+
private static String generateAttributeSortedString(
286+
final PipeParameters pipeConnectorParameters) {
272287
final TreeMap<String, String> sortedStringSourceMap =
273288
new TreeMap<>(pipeConnectorParameters.getAttribute());
274289
sortedStringSourceMap.remove(SystemConstant.RESTART_OR_NEWLY_ADDED_KEY);
275290
return sortedStringSourceMap.toString();
276291
}
277292

293+
/** Masked attribute string for logs, metrics and exception messages. */
294+
private static String generateAttributeDisplayString(
295+
final PipeParameters pipeConnectorParameters) {
296+
final TreeMap<String, String> filteredAttributes =
297+
new TreeMap<>(pipeConnectorParameters.getAttribute());
298+
filteredAttributes.remove(SystemConstant.RESTART_OR_NEWLY_ADDED_KEY);
299+
return new PipeParameters(filteredAttributes).toString();
300+
}
301+
302+
private void throwNoSuchSubtaskException(final String attributeSortedString) {
303+
throw new PipeException(
304+
FAILED_TO_DEREGISTER_EXCEPTION_MESSAGE
305+
+ getDisplayStringForException(attributeSortedString));
306+
}
307+
308+
private String getDisplayStringForException(final String attributeSortedString) {
309+
return attributeSortedString2DisplayString.getOrDefault(attributeSortedString, "unknown");
310+
}
311+
278312
///////////////////////// Singleton Instance Holder /////////////////////////
279313

280314
private PipeSinkSubtaskManager() {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -199,8 +199,8 @@ private void createRate(final String taskID) {
199199

200200
private void createTimer(final String taskID) {
201201
final PipeSinkSubtask sink = sinkMap.get(taskID);
202-
compressionTimerMap.putIfAbsent(
203-
sink.getAttributeSortedString(),
202+
compressionTimerMap.put(
203+
taskID,
204204
metricService.getOrCreateTimer(
205205
Metric.PIPE_COMPRESSION_TIME.toString(),
206206
MetricLevel.IMPORTANT,
@@ -394,7 +394,7 @@ private void removeTimer(final String taskID) {
394394
sink.getAttributeSortedString(),
395395
Tag.CREATION_TIME.toString(),
396396
String.valueOf(sink.getCreationTime()));
397-
compressionTimerMap.remove(sink.getAttributeSortedString());
397+
compressionTimerMap.remove(taskID);
398398
}
399399

400400
private void removeHistogram(final String taskID) {
@@ -492,8 +492,8 @@ public void markPipeHeartbeatEvent(final String taskID) {
492492
rate.mark();
493493
}
494494

495-
public Timer getCompressionTimer(final String attributeSortedString) {
496-
return Objects.isNull(metricService) ? null : compressionTimerMap.get(attributeSortedString);
495+
public Timer getCompressionTimer(final String taskID) {
496+
return Objects.isNull(metricService) ? null : compressionTimerMap.get(taskID);
497497
}
498498

499499
//////////////////////////// singleton ////////////////////////////

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSink.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -604,9 +604,8 @@ protected byte[] getTransferMultiFilePieceBytes(
604604

605605
@Override
606606
protected byte[] compressIfNeeded(final byte[] reqInBytes) throws IOException {
607-
if (Objects.isNull(compressionTimer) && Objects.nonNull(attributeSortedString)) {
608-
compressionTimer =
609-
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(attributeSortedString);
607+
if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
608+
compressionTimer = PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
610609
}
611610
return super.compressIfNeeded(reqInBytes);
612611
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -508,9 +508,8 @@ private void transferBatchedEventsIfNecessary() throws IOException, WriteProcess
508508

509509
@Override
510510
public TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws IOException {
511-
if (Objects.isNull(compressionTimer) && Objects.nonNull(attributeSortedString)) {
512-
compressionTimer =
513-
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(attributeSortedString);
511+
if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
512+
compressionTimer = PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
514513
}
515514
return super.compressIfNeeded(req);
516515
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -595,9 +595,8 @@ private void doTransfer(
595595

596596
@Override
597597
public TPipeTransferReq compressIfNeeded(final TPipeTransferReq req) throws IOException {
598-
if (Objects.isNull(compressionTimer) && Objects.nonNull(attributeSortedString)) {
599-
compressionTimer =
600-
PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(attributeSortedString);
598+
if (Objects.isNull(compressionTimer) && Objects.nonNull(sinkTaskId)) {
599+
compressionTimer = PipeDataRegionSinkMetrics.getInstance().getCompressionTimer(sinkTaskId);
601600
}
602601
return super.compressIfNeeded(req);
603602
}

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/plugin/env/PipeTaskSinkRuntimeEnvironment.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
public class PipeTaskSinkRuntimeEnvironment extends PipeTaskRuntimeEnvironment {
2323
private String attributeSortedString;
24+
private String sinkTaskId;
2425

2526
public PipeTaskSinkRuntimeEnvironment(
2627
final String pipeName, final long creationTime, final int regionId) {
@@ -34,4 +35,12 @@ public String getAttributeSortedString() {
3435
public void setAttributeSortedString(String attributeSortedString) {
3536
this.attributeSortedString = attributeSortedString;
3637
}
38+
39+
public String getSinkTaskId() {
40+
return sinkTaskId;
41+
}
42+
43+
public void setSinkTaskId(final String sinkTaskId) {
44+
this.sinkTaskId = sinkTaskId;
45+
}
3746
}

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBSink.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -189,6 +189,7 @@ public abstract class IoTDBSink implements PipeConnector, PipeConnectorWithEvent
189189
private final AtomicLong totalUncompressedSize = new AtomicLong(0);
190190
private final AtomicLong totalCompressedSize = new AtomicLong(0);
191191
protected String attributeSortedString;
192+
protected String sinkTaskId;
192193
protected Timer compressionTimer;
193194
protected boolean isRealtimeFirst;
194195

@@ -391,8 +392,10 @@ public void customize(
391392
throws Exception {
392393
final PipeRuntimeEnvironment environment = configuration.getRuntimeEnvironment();
393394
if (environment instanceof PipeTaskSinkRuntimeEnvironment) {
394-
attributeSortedString =
395-
((PipeTaskSinkRuntimeEnvironment) environment).getAttributeSortedString();
395+
final PipeTaskSinkRuntimeEnvironment sinkEnvironment =
396+
(PipeTaskSinkRuntimeEnvironment) environment;
397+
attributeSortedString = sinkEnvironment.getAttributeSortedString();
398+
sinkTaskId = sinkEnvironment.getSinkTaskId();
396399
}
397400

398401
nodeUrls.clear();

0 commit comments

Comments
 (0)