You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "FMX (via GitHub)" <gi...@apache.org> on 2023/12/06 12:44:43 UTC
[PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
FMX opened a new pull request, #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135
### What changes were proposed in this pull request?
1. To support io encryption for spark.
### Why are the changes needed?
Ditto.
### Does this PR introduce _any_ user-facing change?
NO.
### How was this patch tested?
GA.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1430838466
##########
client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java:
##########
@@ -578,6 +627,23 @@ private boolean fillBuffer() throws IOException {
} else {
limit = size;
}
+
+ if (decipher != null) {
+ byte[] decryptBuf = new byte[limit];
+ int decryptLength = CryptoUtils.decrypt(decipher, rawDataBuf, 0, limit, decryptBuf);
+ logger.debug(
+ "fetch data decryption shuffleKey: {}, mapId: {}, attempId: {}, batchId: {}, decryptLength/originLength: {}/{}",
+ shuffleKey,
+ mapId,
+ attemptId,
+ batchId,
+ decryptLength,
+ limit);
+ limit = decryptLength;
+ System.arraycopy(decryptBuf, 0, rawDataBuf, 0, limit);
+ decryptBuf = null;
Review Comment:
Couple of comments:
* Instead of deciphering from `rawDataBuf` into `decryptBuf` and then copying the result back to `rawDataBuf` - we can avoid one copy and make the final result go to `rawDataBuf` (and have the input come from another `byte[]` when encryption is enabled - similar to for compression).
* ~If `decryptLength` > `rawDataBuf.length`, the `arraycopy` will fail.~
* On rechecking, this should not happen
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#issuecomment-1862069622
Looks like it got merged already :-)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1432214552
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -1754,4 +1799,19 @@ private boolean connectFail(String message) {
public TransportClientFactory getDataClientFactory() {
return dataClientFactory;
}
+
+ @Override
+ public void setupIoCrypto(
+ Optional<byte[]> ioCryptoKey, Properties ioCryptoConf, byte[] ioCryptoInitializationVector) {
+ this.ioCryptoKey = ioCryptoKey;
+ this.ioCryptoConf = ioCryptoConf;
+ this.ioCyrptoInitializationVector = ioCryptoInitializationVector;
+ if (this.ioCryptoKey.isPresent()) {
+ try {
+ CryptoUtils.getEncipher(this.ioCryptoKey, this.ioCryptoConf, ioCryptoInitializationVector);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to init encipher", e);
Review Comment:
Initialize encipher failed does means that the argements are not correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#issuecomment-1842834459
## [Codecov](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2135?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: `4 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`406cef8`)](https://app.codecov.io/gh/apache/incubator-celeborn/commit/406cef83925c8628c0d6bd6e5628cc7b47534ff2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 47.20% compared to head [(`0ea77a6`)](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2135?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 46.30%.
> Report is 2 commits behind head on main.
> :exclamation: Current head 0ea77a6 differs from pull request most recent head 075f544. Consider uploading reports for the commit 075f544 to get more accurate results
| [Files](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2135?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...cala/org/apache/celeborn/common/CelebornConf.scala](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2135?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL0NlbGVib3JuQ29uZi5zY2FsYQ==) | 83.34% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2135?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #2135 +/- ##
==========================================
- Coverage 47.20% 46.30% -0.89%
==========================================
Files 173 175 +2
Lines 11039 11357 +318
Branches 1011 1016 +5
==========================================
+ Hits 5210 5258 +48
- Misses 5486 5751 +265
- Partials 343 348 +5
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2135?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1429681975
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java:
##########
@@ -139,6 +169,16 @@ private void initializeLifecycleManager() {
synchronized (this) {
if (lifecycleManager == null) {
lifecycleManager = new LifecycleManager(appUniqueId, celebornConf);
+ shuffleClient =
Review Comment:
We don't need to initialize the client on the driver
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX closed pull request #2135: [CELEBORN-1150] support io encryption for spark
URL: https://github.com/apache/incubator-celeborn/pull/2135
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#issuecomment-1863754637
> Looks like it got merged already :-)
Thanks, I will come up with a followup to address your comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1432252706
##########
client/src/main/java/org/apache/celeborn/client/security/CryptoUtils.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client.security;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.ShortBufferException;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+
+import org.apache.commons.crypto.cipher.CryptoCipher;
+import org.apache.commons.crypto.random.CryptoRandomFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CryptoUtils {
+ private static Logger logger = LoggerFactory.getLogger(CryptoUtils.class);
+ public static final int IV_LENGTH_IN_BYTES = 16;
+ public static final String COMMONS_CRYPTO_CONFIG_PREFIX = "commons.crypto.";
+ public static final String COMMONS_CRYPTO_CONFIG_TRANSFORMATION =
+ COMMONS_CRYPTO_CONFIG_PREFIX + "cipher.transformation";
+ public static final String CRYPTO_ALGORITHM = "AES";
+
+ public static byte[] createIoCryptoInitializationVector() {
+ byte[] iv = new byte[IV_LENGTH_IN_BYTES];
+ long initialIVStart = System.nanoTime();
+ try {
+ CryptoRandomFactory.getCryptoRandom(new Properties()).nextBytes(iv);
+ } catch (GeneralSecurityException e) {
+ logger.warn("Failed to create crypto Initialization Vector", e);
+ iv = "1234567890123456".getBytes();
+ }
+ long initialIVFinish = System.nanoTime();
+ long initialIVTime = TimeUnit.NANOSECONDS.toMillis(initialIVFinish - initialIVStart);
+ if (initialIVTime > 2000) {
+ logger.warn(
+ "It costs {} milliseconds to create the Initialization Vector used by crypto",
+ initialIVTime);
Review Comment:
Given this is an infrequent operation, I was trying to understand what is the concern we are measuring for.
Are we expecting delays with `CryptoRandomFactory.getCryptoRandom.nextBytes` ? When/how are we expecting it to occur (taking `>= 2s`) ?
Trying to understand the scenario where delays would be observed - in other words, if this warning is emitted to the logs, what would be the reason for triggering it ? And how can developers/users troubleshoot/mitigate it ? Thanks !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1430828180
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -157,6 +159,29 @@ public void update(ReduceFileGroups fileGroups) {
protected final Map<Integer, ReduceFileGroups> reduceFileGroupsMap =
JavaUtils.newConcurrentHashMap();
+ protected Optional<byte[]> ioCryptoKey = Optional.empty();
+
+ protected Properties ioCryptoConf;
+
+ protected byte[] ioCyrptoInitializationVector;
+
+ private ThreadLocal<CryptoCipher> encipherThreadLocal =
+ new ThreadLocal<CryptoCipher>() {
+ @Override
+ protected CryptoCipher initialValue() {
+ CryptoCipher cryptoCipher = null;
+ if (ioCryptoKey.isPresent()) {
Review Comment:
nit: We dont need the `ioCryptoKey.isPresent()` - we can make it a `Precondition`
##########
client/src/main/java/org/apache/celeborn/client/security/CryptoUtils.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client.security;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.ShortBufferException;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+
+import org.apache.commons.crypto.cipher.CryptoCipher;
+import org.apache.commons.crypto.random.CryptoRandomFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CryptoUtils {
+ private static Logger logger = LoggerFactory.getLogger(CryptoUtils.class);
+ public static final int IV_LENGTH_IN_BYTES = 16;
+ public static final String COMMONS_CRYPTO_CONFIG_PREFIX = "commons.crypto.";
+ public static final String COMMONS_CRYPTO_CONFIG_TRANSFORMATION =
+ COMMONS_CRYPTO_CONFIG_PREFIX + "cipher.transformation";
+ public static final String CRYPTO_ALGORITHM = "AES";
+
+ public static byte[] createIoCryptoInitializationVector() {
+ byte[] iv = new byte[IV_LENGTH_IN_BYTES];
+ long initialIVStart = System.nanoTime();
+ try {
+ CryptoRandomFactory.getCryptoRandom(new Properties()).nextBytes(iv);
+ } catch (GeneralSecurityException e) {
+ logger.warn("Failed to create crypto Initialization Vector", e);
+ iv = "1234567890123456".getBytes();
Review Comment:
This points to invalid config - in which case, fail instead ?
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -878,6 +903,23 @@ public int pushOrMergeData(
// increment batchId
final int nextBatchId = pushState.nextBatchId();
+ if (ioCryptoKey.isPresent()) {
+ CryptoCipher encipher = encipherThreadLocal.get();
+ byte[] encryptData = new byte[length + encipher.getBlockSize()];
+ int encryptLength = CryptoUtils.encrypt(encipher, data, offset, length, encryptData);
Review Comment:
QQ: given we are doing update and final in `encrypt`, do we need to set the size to `length + blockSize * 2` ?
I am blanking out on what the expectation is - please check.
##########
client/src/main/java/org/apache/celeborn/client/security/CryptoUtils.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client.security;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.ShortBufferException;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+
+import org.apache.commons.crypto.cipher.CryptoCipher;
+import org.apache.commons.crypto.random.CryptoRandomFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CryptoUtils {
+ private static Logger logger = LoggerFactory.getLogger(CryptoUtils.class);
+ public static final int IV_LENGTH_IN_BYTES = 16;
+ public static final String COMMONS_CRYPTO_CONFIG_PREFIX = "commons.crypto.";
+ public static final String COMMONS_CRYPTO_CONFIG_TRANSFORMATION =
+ COMMONS_CRYPTO_CONFIG_PREFIX + "cipher.transformation";
+ public static final String CRYPTO_ALGORITHM = "AES";
+
+ public static byte[] createIoCryptoInitializationVector() {
+ byte[] iv = new byte[IV_LENGTH_IN_BYTES];
+ long initialIVStart = System.nanoTime();
+ try {
+ CryptoRandomFactory.getCryptoRandom(new Properties()).nextBytes(iv);
+ } catch (GeneralSecurityException e) {
+ logger.warn("Failed to create crypto Initialization Vector", e);
+ iv = "1234567890123456".getBytes();
+ }
+ long initialIVFinish = System.nanoTime();
+ long initialIVTime = TimeUnit.NANOSECONDS.toMillis(initialIVFinish - initialIVStart);
+ if (initialIVTime > 2000) {
+ logger.warn(
+ "It costs {} milliseconds to create the Initialization Vector used by crypto",
+ initialIVTime);
Review Comment:
QQ: Is this a concern ? When/How would this occur ?
##########
client-spark/spark-2/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -18,6 +18,7 @@
package org.apache.spark.shuffle.celeborn
import java.io.IOException
+import java.util.{Optional, Properties}
Review Comment:
revert ?
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -4240,4 +4246,28 @@ object CelebornConf extends Logging {
.version("0.4.0")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("30s")
+
+ val SPARK_CLIENT_IO_ENCRYPTION_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.spark.io.encryption.enabled")
+ .categories("client")
+ .version("0.4.0")
+ .doc("whether to enable io encryption")
+ .booleanConf
+ .createWithDefault(true)
+
+ val SPARK_CLIENT_IO_ENCRYPTION_KEY: ConfigEntry[String] =
+ buildConf("celeborn.client.spark.io.encryption.key")
+ .categories("client")
+ .version("0.4.0")
+ .doc("io encryption key")
+ .stringConf
+ .createWithDefault("")
+
+ val SPARK_CLIENT_IO_ENCRYPTION_INITIALIZATION_VECTOR: ConfigEntry[String] =
+ buildConf("celeborn.client.spark.io.encryption.initialization.vector")
+ .categories("client")
+ .version("0.4.0")
+ .doc("io encryption initialization vector")
+ .stringConf
+ .createWithDefault("")
Review Comment:
We can drop:
* `SPARK_CLIENT_IO_ENCRYPTION_KEY`
* `SPARK_CLIENT_IO_ENCRYPTION_INITIALIZATION_VECTOR`
* `sparkIoEncryptionKey`
* `sparkIoEncryptionInitializationVector`
They are not being used.
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala:
##########
@@ -59,6 +59,8 @@ trait SparkTestBase extends AnyFunSuite
sparkConf.set("spark.sql.adaptive.localShuffleReader.enabled", "false")
sparkConf.set(s"spark.${MASTER_ENDPOINTS.key}", masterInfo._1.rpcEnv.address.toString)
sparkConf.set(s"spark.${SPARK_SHUFFLE_WRITER_MODE.key}", mode.toString)
+ sparkConf.set("spark.io.encryption.enabled", "true")
+ sparkConf.set("spark.io.crypto.cipher.transformation", "AES/CBC/PKCS5Padding")
Review Comment:
The tests should check for both cases - with and without io encryption enabled
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -1754,4 +1799,19 @@ private boolean connectFail(String message) {
public TransportClientFactory getDataClientFactory() {
return dataClientFactory;
}
+
+ @Override
+ public void setupIoCrypto(
+ Optional<byte[]> ioCryptoKey, Properties ioCryptoConf, byte[] ioCryptoInitializationVector) {
+ this.ioCryptoKey = ioCryptoKey;
+ this.ioCryptoConf = ioCryptoConf;
+ this.ioCyrptoInitializationVector = ioCryptoInitializationVector;
+ if (this.ioCryptoKey.isPresent()) {
+ try {
+ CryptoUtils.getEncipher(this.ioCryptoKey, this.ioCryptoConf, ioCryptoInitializationVector);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to init encipher", e);
Review Comment:
`RuntimeException` -> `IllegalArgumentException` ?
##########
client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java:
##########
@@ -578,6 +627,23 @@ private boolean fillBuffer() throws IOException {
} else {
limit = size;
}
+
+ if (decipher != null) {
+ byte[] decryptBuf = new byte[limit];
+ int decryptLength = CryptoUtils.decrypt(decipher, rawDataBuf, 0, limit, decryptBuf);
+ logger.debug(
+ "fetch data decryption shuffleKey: {}, mapId: {}, attempId: {}, batchId: {}, decryptLength/originLength: {}/{}",
+ shuffleKey,
+ mapId,
+ attemptId,
+ batchId,
+ decryptLength,
+ limit);
+ limit = decryptLength;
+ System.arraycopy(decryptBuf, 0, rawDataBuf, 0, limit);
+ decryptBuf = null;
Review Comment:
Couple of comments:
* Instead of deciphering from `rawDataBuf` into `decryptBuf` and then copying the result back to `rawDataBuf` - we can avoid one copy and make the final result go to `rawDataBuf` (and have the input come from another `byte[]` when encryption is enabled - similar to for compression).
* If `decryptLength` > `rawDataBuf.length`, the `arraycopy` will fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1429705087
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -758,6 +758,12 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def clientExcludedWorkerExpireTimeout: Long = get(CLIENT_EXCLUDED_WORKER_EXPIRE_TIMEOUT)
def clientExcludeReplicaOnFailureEnabled: Boolean =
get(CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED)
+ def ioEncryptionEnabled: Boolean = get(CLIENT_IO_ENCRYPTION_ENABLED)
+
Review Comment:
nit: remove the blank lines between the 3 encrypt related method
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkUtils.java:
##########
@@ -89,6 +91,10 @@ public static CelebornConf fromSparkConf(SparkConf conf) {
}
public static String appUniqueId(SparkContext context) {
+ return appUniqueId(context, fromSparkConf(context.conf()));
+ }
+
+ public static String appUniqueId(SparkContext context, CelebornConf celebornConf) {
Review Comment:
Seems this method is never used.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -4240,4 +4246,28 @@ object CelebornConf extends Logging {
.version("0.4.0")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("30s")
+
+ val CLIENT_IO_ENCRYPTION_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.io.encryption.enabled")
Review Comment:
0.4.0
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -4240,4 +4246,28 @@ object CelebornConf extends Logging {
.version("0.4.0")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("30s")
+
+ val CLIENT_IO_ENCRYPTION_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.io.encryption.enabled")
+ .categories("client")
+ .version("0.3.0")
+ .doc("whether to enable io encryption")
+ .booleanConf
+ .createWithDefault(true)
+
+ val CLIENT_IO_ENCRYPTION_KEY: ConfigEntry[String] =
+ buildConf("celeborn.client.io.encryption.key")
Review Comment:
ditto
##########
client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkUtils.java:
##########
@@ -112,11 +112,17 @@ public static CelebornConf fromSparkConf(SparkConf conf) {
}
public static String appUniqueId(SparkContext context) {
+ return appUniqueId(context, fromSparkConf(context.conf()));
+ }
+
+ public static String appUniqueId(SparkContext context, CelebornConf celebornConf) {
+ String appUniqueId = "";
if (context.applicationAttemptId().isDefined()) {
Review Comment:
ditto
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -4240,4 +4246,28 @@ object CelebornConf extends Logging {
.version("0.4.0")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("30s")
+
+ val CLIENT_IO_ENCRYPTION_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.io.encryption.enabled")
Review Comment:
Better use celeborn.client.spark.io.encryption.enabled
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -4240,4 +4246,28 @@ object CelebornConf extends Logging {
.version("0.4.0")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("30s")
+
+ val CLIENT_IO_ENCRYPTION_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.io.encryption.enabled")
+ .categories("client")
+ .version("0.3.0")
+ .doc("whether to enable io encryption")
+ .booleanConf
+ .createWithDefault(true)
+
+ val CLIENT_IO_ENCRYPTION_KEY: ConfigEntry[String] =
+ buildConf("celeborn.client.io.encryption.key")
+ .categories("client")
+ .version("0.3.0")
+ .doc("io encryption key")
+ .stringConf
+ .createWithDefault("")
+
+ val CLIENT_IO_ENCRYPTION_INITIALIZATION_VECTOR: ConfigEntry[String] =
+ buildConf("celeborn.client.io.encryption.initialization.vector")
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#issuecomment-1844153611
> +CC @otterc who is working on adding TLS support as part of SASL authentication.
>
> Do we want to include custom encryption in Celeborn ? Spark 4 has also added support for TLS, and I would expect that to become the default in a future version of spark as well.
Hi @mridulm, this patch will follow Spark's encryption and description mechanism for shuffle data, not for the network link.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1432216034
##########
client/src/main/java/org/apache/celeborn/client/security/CryptoUtils.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client.security;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.ShortBufferException;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+
+import org.apache.commons.crypto.cipher.CryptoCipher;
+import org.apache.commons.crypto.random.CryptoRandomFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CryptoUtils {
+ private static Logger logger = LoggerFactory.getLogger(CryptoUtils.class);
+ public static final int IV_LENGTH_IN_BYTES = 16;
+ public static final String COMMONS_CRYPTO_CONFIG_PREFIX = "commons.crypto.";
+ public static final String COMMONS_CRYPTO_CONFIG_TRANSFORMATION =
+ COMMONS_CRYPTO_CONFIG_PREFIX + "cipher.transformation";
+ public static final String CRYPTO_ALGORITHM = "AES";
+
+ public static byte[] createIoCryptoInitializationVector() {
+ byte[] iv = new byte[IV_LENGTH_IN_BYTES];
+ long initialIVStart = System.nanoTime();
+ try {
+ CryptoRandomFactory.getCryptoRandom(new Properties()).nextBytes(iv);
+ } catch (GeneralSecurityException e) {
+ logger.warn("Failed to create crypto Initialization Vector", e);
+ iv = "1234567890123456".getBytes();
+ }
+ long initialIVFinish = System.nanoTime();
+ long initialIVTime = TimeUnit.NANOSECONDS.toMillis(initialIVFinish - initialIVStart);
+ if (initialIVTime > 2000) {
+ logger.warn(
+ "It costs {} milliseconds to create the Initialization Vector used by crypto",
+ initialIVTime);
Review Comment:
Just to record an abnormal situation. It might happen if a driver/executor is running on a busy server.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#issuecomment-1858652854
+CC @otterc , looks like I wont have time to review this unfortunately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1432250542
##########
client-spark/spark-2/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -18,6 +18,7 @@
package org.apache.spark.shuffle.celeborn
import java.io.IOException
+import java.util.{Optional, Properties}
Review Comment:
Not in this file, right ? The only diff here is the imports.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#issuecomment-1843537842
+CC @otterc who is working on adding TLS support as part of SASL authentication.
Do we want to include custom encryption in Celeborn ? Spark 4 has also added support for TLS, and I would expect that to become the default in a future version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#issuecomment-1844446856
Ah, I misunderstood.
If this is equivalent to spark's `spark.io.encryption.enabled`, we should avoid the limitations Spark has with that config.
(You can search for usages of `IO_ENCRYPTION_ENABLED` in spark codebase).
Will go over this draft later this week, thanks for working on it !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1432252706
##########
client/src/main/java/org/apache/celeborn/client/security/CryptoUtils.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client.security;
+
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.ShortBufferException;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+
+import org.apache.commons.crypto.cipher.CryptoCipher;
+import org.apache.commons.crypto.random.CryptoRandomFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CryptoUtils {
+ private static Logger logger = LoggerFactory.getLogger(CryptoUtils.class);
+ public static final int IV_LENGTH_IN_BYTES = 16;
+ public static final String COMMONS_CRYPTO_CONFIG_PREFIX = "commons.crypto.";
+ public static final String COMMONS_CRYPTO_CONFIG_TRANSFORMATION =
+ COMMONS_CRYPTO_CONFIG_PREFIX + "cipher.transformation";
+ public static final String CRYPTO_ALGORITHM = "AES";
+
+ public static byte[] createIoCryptoInitializationVector() {
+ byte[] iv = new byte[IV_LENGTH_IN_BYTES];
+ long initialIVStart = System.nanoTime();
+ try {
+ CryptoRandomFactory.getCryptoRandom(new Properties()).nextBytes(iv);
+ } catch (GeneralSecurityException e) {
+ logger.warn("Failed to create crypto Initialization Vector", e);
+ iv = "1234567890123456".getBytes();
+ }
+ long initialIVFinish = System.nanoTime();
+ long initialIVTime = TimeUnit.NANOSECONDS.toMillis(initialIVFinish - initialIVStart);
+ if (initialIVTime > 2000) {
+ logger.warn(
+ "It costs {} milliseconds to create the Initialization Vector used by crypto",
+ initialIVTime);
Review Comment:
Given this is an infrequent operation, I was trying to understand what is the concern we are measuring for.
Are we expecting delays with `CryptoRandomFactory.getCryptoRandom.nextBytes` ? When/how are we expecting it to occur ?
Trying to understand the scenario where delays would be observed - in other words, if this warning is emitted to the logs, what would be the reason for triggering it ? And how can developers/users troubleshoot/mitigate it ? Thanks !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1150] support io encryption for spark [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #2135:
URL: https://github.com/apache/incubator-celeborn/pull/2135#discussion_r1432213595
##########
client-spark/spark-2/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala:
##########
@@ -18,6 +18,7 @@
package org.apache.spark.shuffle.celeborn
import java.io.IOException
+import java.util.{Optional, Properties}
Review Comment:
This PR is using java.util.Optional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org