You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uniffle.apache.org by xi...@apache.org on 2023/11/20 07:28:18 UTC
(incubator-uniffle) branch branch-0.8 updated (81396624d -> cf25897a7)
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a change to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
from 81396624d Revert "[#1309] fix(spark): WriteBufferManager in Spark2 does not use a reassigned shuffle server. (#1310)"
new 3904cbdec [MINOR] test: flaky test GrpcServerTest.testGrpcExecutorPool (#1321)
new 42b5c21d3 [MINOR] fix: flaky test ShuffleTaskManagerTest#checkAndClearLeakShuffleDataTest (#1320)
new cf25897a7 [#1290] improvement(operator): Avoid accidentally deleting data of other services when misconfiguring the mounting directory (#1291)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/uniffle/common/rpc/GrpcServer.java | 7 +++
.../apache/uniffle/common/rpc/GrpcServerTest.java | 1 +
.../operator/pkg/controller/constants/constants.go | 3 ++
.../controller/sync/shuffleserver/shuffleserver.go | 2 +-
.../operator/pkg/controller/util/util.go | 3 +-
.../operator/pkg/controller/util/util_test.go | 51 ++++++++++++++++++++++
.../uniffle/server/ShuffleTaskManagerTest.java | 5 ++-
7 files changed, 68 insertions(+), 4 deletions(-)
(incubator-uniffle) 03/03: [#1290] improvement(operator): Avoid accidentally deleting data of other services when misconfiguring the mounting directory (#1291)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit cf25897a72f4a6cde34dccc5f5f7844a410b545a
Author: QI Jiale <qi...@foxmail.com>
AuthorDate: Sat Nov 18 14:41:56 2023 +0800
[#1290] improvement(operator): Avoid accidentally deleting data of other services when misconfiguring the mounting directory (#1291)
### What changes were proposed in this pull request?
Only delete `rssdata` directory.
### Why are the changes needed?
Fix: #1290
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Tested in our cluster.
1. Mount `/data/hdfs1/rssdata1` dir for shuffle server，the server will create `/data/hdfs1/rssdata1/rssdata/`
2. Create files manually at host machine
```
touch /data/hdfs1/rssdata1/a.txt
touch /data/hdfs1/rssdata1/rssdata/b.txt
```
3. Update crd to terminate this shuffle server.
4. Without this pr, both `a.txt`, `b.txt` and `rssdata` dir are deleted
5. With this pr, only `b.txt` is deleted
Co-authored-by: 齐家乐(26731624) <qi...@ke.com>
(cherry picked from commit be106976474cc02a614ca842421670debf50f7aa)
---
.../operator/pkg/controller/constants/constants.go | 3 ++
.../controller/sync/shuffleserver/shuffleserver.go | 2 +-
.../operator/pkg/controller/util/util.go | 3 +-
.../operator/pkg/controller/util/util_test.go | 51 ++++++++++++++++++++++
4 files changed, 57 insertions(+), 2 deletions(-)
diff --git a/deploy/kubernetes/operator/pkg/controller/constants/constants.go b/deploy/kubernetes/operator/pkg/controller/constants/constants.go
index 8bda1ca6b..41c0f34c0 100644
--- a/deploy/kubernetes/operator/pkg/controller/constants/constants.go
+++ b/deploy/kubernetes/operator/pkg/controller/constants/constants.go
@@ -58,6 +58,9 @@ const (
// ConfigurationVolumeName is the name of configMap volume records configuration of coordinators or shuffle servers.
ConfigurationVolumeName = "configuration"
+
+ //RssDataDir is the directory name for RSS data as the local storage.
+ RssDataDir = "rssdata"
)
// PropertyKey defines property key in configuration of coordinators or shuffle servers.
diff --git a/deploy/kubernetes/operator/pkg/controller/sync/shuffleserver/shuffleserver.go b/deploy/kubernetes/operator/pkg/controller/sync/shuffleserver/shuffleserver.go
index 7b4baad69..6404456ba 100644
--- a/deploy/kubernetes/operator/pkg/controller/sync/shuffleserver/shuffleserver.go
+++ b/deploy/kubernetes/operator/pkg/controller/sync/shuffleserver/shuffleserver.go
@@ -236,7 +236,7 @@ func generateStorageBasePath(rss *unifflev1alpha1.RemoteShuffleService) string {
if k == rss.Spec.ShuffleServer.LogHostPath {
continue
}
- paths = append(paths, strings.TrimSuffix(v, "/")+"/rssdata")
+ paths = append(paths, strings.TrimSuffix(v, "/")+"/"+controllerconstants.RssDataDir)
}
sort.Strings(paths)
return strings.Join(paths, ",")
diff --git a/deploy/kubernetes/operator/pkg/controller/util/util.go b/deploy/kubernetes/operator/pkg/controller/util/util.go
index 99cb0aff8..5a1375f43 100644
--- a/deploy/kubernetes/operator/pkg/controller/util/util.go
+++ b/deploy/kubernetes/operator/pkg/controller/util/util.go
@@ -86,7 +86,8 @@ func addVolumeMountsOfMainContainer(mainContainer *corev1.Container,
mainContainer.VolumeMounts = append(mainContainer.VolumeMounts,
GenerateHostPathVolumeMounts(hostPathMounts)...)
for _, mountPath := range hostPathMounts {
- clearPathCMDs = append(clearPathCMDs, fmt.Sprintf("rm -rf %v/*", strings.TrimSuffix(mountPath, "/")))
+ clearPathCMDs = append(clearPathCMDs, fmt.Sprintf("rm -rf %v/%v/*",
+ strings.TrimSuffix(mountPath, "/"), controllerconstants.RssDataDir))
}
if len(clearPathCMDs) > 0 {
mainContainer.Lifecycle = &corev1.Lifecycle{
diff --git a/deploy/kubernetes/operator/pkg/controller/util/util_test.go b/deploy/kubernetes/operator/pkg/controller/util/util_test.go
index 864d8ee2d..d7e6d7a14 100644
--- a/deploy/kubernetes/operator/pkg/controller/util/util_test.go
+++ b/deploy/kubernetes/operator/pkg/controller/util/util_test.go
@@ -19,6 +19,7 @@ package util
import (
"sort"
+ "strings"
"testing"
"github.com/stretchr/testify/assert"
@@ -179,6 +180,56 @@ func TestAddOwnerReference(t *testing.T) {
}
}
+func TestAddVolumeMountsOfMainContainer(t *testing.T) {
+ for _, tt := range []struct {
+ name string
+ mainContainer *corev1.Container
+ hostPathMounts map[string]string
+ volumeMounts []corev1.VolumeMount
+ expectedCommand string
+ }{
+ {
+ name: "check add volume mount 1",
+ mainContainer: &corev1.Container{},
+ hostPathMounts: map[string]string{
+ "/rssdata1/data1": "/data1",
+ "/rssdata2/data2": "/data2",
+ "/rssdata3/data2": "/data3",
+ },
+ volumeMounts: []corev1.VolumeMount{},
+ expectedCommand: "rm -rf /data1/rssdata/*;rm -rf /data2/rssdata/*;rm -rf /data3/rssdata/*",
+ },
+ {
+ name: "check add volume mount 2",
+ mainContainer: &corev1.Container{},
+ hostPathMounts: map[string]string{
+ "/rssdata4/data4": "/data4",
+ "/rssdata5/data5": "/data5",
+ "/rssdata6/data6": "/data6",
+ },
+ volumeMounts: []corev1.VolumeMount{
+ {
+ Name: "/rsslog",
+ MountPath: "/rsslog",
+ },
+ },
+ expectedCommand: "rm -rf /data4/rssdata/*;rm -rf /data5/rssdata/*;rm -rf /data6/rssdata/*",
+ },
+ } {
+ t.Run(tt.name, func(t *testing.T) {
+ assertion := assert.New(t)
+ expectedVolumeMountCount := len(tt.hostPathMounts) + len(tt.volumeMounts)
+ assertion.Equal(0, len(tt.mainContainer.VolumeMounts))
+ addVolumeMountsOfMainContainer(tt.mainContainer, tt.hostPathMounts, tt.volumeMounts)
+ assertion.Equal(expectedVolumeMountCount, len(tt.mainContainer.VolumeMounts))
+ strSlice := strings.Split(tt.mainContainer.Lifecycle.PreStop.Exec.Command[2], ";")
+ sort.Strings(strSlice)
+ sortedString := strings.Join(strSlice, ";")
+ assertion.Equal(tt.expectedCommand, sortedString)
+ })
+ }
+}
+
func buildRssWithUID() *uniffleapi.RemoteShuffleService {
rss := utils.BuildRSSWithDefaultValue()
rss.UID = "uid-test"
(incubator-uniffle) 01/03: [MINOR] test: flaky test GrpcServerTest.testGrpcExecutorPool (#1321)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit 3904cbdec4ded78d89781b2415a0092b8c5d4c00
Author: xianjingfeng <xi...@gmail.com>
AuthorDate: Fri Nov 17 17:00:03 2023 +0800
[MINOR] test: flaky test GrpcServerTest.testGrpcExecutorPool (#1321)
### What changes were proposed in this pull request?
fix flaky test GrpcServerTest.testGrpcExecutorPool
### Why are the changes needed?
GrpcServerTest.testGrpcExecutorPool:83 expected: <1.0> but was: <0.0>
The cause of the problem:
The metric GRPC_SERVER_EXECUTOR_BLOCKING_QUEUE_SIZE_KEY is updated only when there is thread be executed. So if the first two threads have already started executing, and then the third thread is submitted, this metric will not be updated.
### Does this PR introduce any user-facing change?
No.
### How was this patch tested?
Repeat it for 100 times in my development enviroment.
(cherry picked from commit 60e3be8591e2bf5b23e150f12e18821fcded1adc)
---
common/src/main/java/org/apache/uniffle/common/rpc/GrpcServer.java | 7 +++++++
.../test/java/org/apache/uniffle/common/rpc/GrpcServerTest.java | 1 +
2 files changed, 8 insertions(+)
diff --git a/common/src/main/java/org/apache/uniffle/common/rpc/GrpcServer.java b/common/src/main/java/org/apache/uniffle/common/rpc/GrpcServer.java
index 38ae7a88f..cd7f88952 100644
--- a/common/src/main/java/org/apache/uniffle/common/rpc/GrpcServer.java
+++ b/common/src/main/java/org/apache/uniffle/common/rpc/GrpcServer.java
@@ -27,6 +27,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Queues;
import io.grpc.BindableService;
@@ -179,6 +180,12 @@ public class GrpcServer implements ServerInterface {
GRPCMetrics.GRPC_SERVER_EXECUTOR_BLOCKING_QUEUE_SIZE_KEY, getQueue().size());
super.afterExecute(r, t);
}
+
+ @VisibleForTesting
+ void correctMetrics() {
+ grpcMetrics.setGauge(
+ GRPCMetrics.GRPC_SERVER_EXECUTOR_BLOCKING_QUEUE_SIZE_KEY, getQueue().size());
+ }
}
@Override
diff --git a/common/src/test/java/org/apache/uniffle/common/rpc/GrpcServerTest.java b/common/src/test/java/org/apache/uniffle/common/rpc/GrpcServerTest.java
index 2142bfcf6..50c4cc9c8 100644
--- a/common/src/test/java/org/apache/uniffle/common/rpc/GrpcServerTest.java
+++ b/common/src/test/java/org/apache/uniffle/common/rpc/GrpcServerTest.java
@@ -75,6 +75,7 @@ public class GrpcServerTest {
Thread.yield();
}
Thread.sleep(120);
+ executor.correctMetrics();
double activeThreads =
grpcMetrics.getGaugeMap().get(GRPC_SERVER_EXECUTOR_ACTIVE_THREADS_KEY).get();
assertEquals(2, activeThreads);
(incubator-uniffle) 02/03: [MINOR] fix: flaky test ShuffleTaskManagerTest#checkAndClearLeakShuffleDataTest (#1320)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit 42b5c21d343ad03a65c9557ffc8bb5f736741829
Author: xianjingfeng <xi...@gmail.com>
AuthorDate: Fri Nov 17 17:00:58 2023 +0800
[MINOR] fix: flaky test ShuffleTaskManagerTest#checkAndClearLeakShuffleDataTest (#1320)
### What changes were proposed in this pull request?
fix flaky test ShuffleTaskManagerTest#checkAndClearLeakShuffleDataTest.
### Why are the changes needed?
ShuffleTaskManagerTest.checkAndClearLeakShuffleDataTest:940 expected:
<false> but was: <true>
The cause of the problem:
Application in shuffleTaskInfos is removed first, and then remove the directory in the local disk.
Refer: org.apache.uniffle.server.ShuffleTaskManager#removeResources
### Does this PR introduce any user-facing change?
No.
### How was this patch tested?
Repeat it for 100 times in my development enviroment.
(cherry picked from commit e1a54999ec2edf280b73e20e76dfcf1d89278d00)
---
.../test/java/org/apache/uniffle/server/ShuffleTaskManagerTest.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/server/src/test/java/org/apache/uniffle/server/ShuffleTaskManagerTest.java b/server/src/test/java/org/apache/uniffle/server/ShuffleTaskManagerTest.java
index 36a089b0e..b87735b37 100644
--- a/server/src/test/java/org/apache/uniffle/server/ShuffleTaskManagerTest.java
+++ b/server/src/test/java/org/apache/uniffle/server/ShuffleTaskManagerTest.java
@@ -936,8 +936,9 @@ public class ShuffleTaskManagerTest extends HadoopTestBase {
File hiddenFile = new File(storageDir + "/" + LocalStorageChecker.CHECKER_DIR_NAME);
hiddenFile.mkdir();
- appIdsOnDisk = getAppIdsOnDisk(localStorageManager);
- assertFalse(appIdsOnDisk.contains(appId));
+ Awaitility.await()
+ .timeout(10, TimeUnit.SECONDS)
+ .until(() -> !getAppIdsOnDisk(localStorageManager).contains(appId));
assertFalse(appIdsOnDisk.contains(LocalStorageChecker.CHECKER_DIR_NAME));
// mock leak shuffle data