Skip to content

Commit 8236b96

Browse files
sanskar-singh-2403sanskar-singh-2403
committed
[Test] Add NetworkPolicy e2e coverage for RayJob and RayService
Follow-up to #4638 / #5030. Adds NetworkPolicy (mode DenyAll) e2e tests that run on the enforcing v1.35 Kind node already used by the Buildkite e2e jobs: - RayJob without clusterSelector using runtimeEnv - RayJob with clusterSelector (explicit submitter ingress rule) - RayService (verifies the Serve app returns a response to an external curl pod) Shared rule builders added in test/support/networkpolicy.go. No CI wiring change needed; tests run inside the existing e2erayjob and e2erayservice Buildkite steps.
1 parent 556d1b4 commit 8236b96

5 files changed

Lines changed: 393 additions & 1 deletion

File tree

‎.buildkite/values-kuberay-operator-override.yaml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,3 +24,5 @@ featureGates:
2424
enabled: true
2525
- name: RayClusterMTLS
2626
enabled: true
27+
- name: RayClusterNetworkPolicy
28+
enabled: true

‎ray-operator/config/overlays/test-overrides/deployment-override.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,4 +9,4 @@ spec:
99
containers:
1010
- name: kuberay-operator
1111
args:
12-
- --feature-gates=RayClusterStatusConditions=true,RayJobDeletionPolicy=true,RayMultiHostIndexing=true,RayCronJob=true,RayServiceIncrementalUpgrade=true,SidecarSubmitterRestart=true,GCSFaultToleranceEmbeddedStorage=true,RayClusterMTLS=true
12+
- --feature-gates=RayClusterStatusConditions=true,RayJobDeletionPolicy=true,RayMultiHostIndexing=true,RayCronJob=true,RayServiceIncrementalUpgrade=true,SidecarSubmitterRestart=true,GCSFaultToleranceEmbeddedStorage=true,RayClusterMTLS=true,RayClusterNetworkPolicy=true

‎ray-operator/test/e2erayjob/rayjob_network_policy_test.go‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
package e2erayjob
2+
3+
import (
4+
"testing"
5+
6+
. "github.com/onsi/gomega"
7+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
8+
9+
rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
10+
"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
11+
rayv1ac "github.com/ray-project/kuberay/ray-operator/pkg/client/applyconfiguration/ray/v1"
12+
. "github.com/ray-project/kuberay/ray-operator/test/support"
13+
)
14+
15+
// RayJob without clusterSelector, under NetworkPolicy DenyAll. The operator owns
16+
// the RayCluster so it adds the submitter ingress rule itself; we only add DNS
17+
// egress. Kindnet enforces NetworkPolicy on k8s >= 1.32 (CI runs on 1.35), so a
18+
// wrong head/worker policy means the pods can't resolve each other and the job
19+
// never finishes.
20+
func TestRayJobWithNetworkPolicy(t *testing.T) {
21+
test := With(t)
22+
g := NewWithT(t)
23+
24+
namespace := test.NewTestNamespace()
25+
26+
// Job scripts.
27+
jobsAC := NewConfigMap(namespace.Name, Files(test, "counter.py"))
28+
jobs, err := test.Client().Core().CoreV1().ConfigMaps(namespace.Name).Apply(test.Ctx(), jobsAC, TestApplyOptions)
29+
g.Expect(err).NotTo(HaveOccurred())
30+
LogWithTimestamp(test.T(), "Created ConfigMap %s/%s successfully", jobs.Namespace, jobs.Name)
31+
32+
// DenyAll: intra-cluster and submitter ingress are handled by the operator,
33+
// we just need DNS egress.
34+
networkPolicy := rayv1ac.NetworkPolicyConfig().
35+
WithMode(rayv1.NetworkPolicyDenyAll).
36+
WithHead(rayv1ac.NetworkPolicyRules().WithEgressRules(DNSEgressRule())).
37+
WithWorker(rayv1ac.NetworkPolicyRules().WithEgressRules(DNSEgressRule()))
38+
39+
rayJobAC := rayv1ac.RayJob("counter", namespace.Name).
40+
WithSpec(rayv1ac.RayJobSpec().
41+
WithRayClusterSpec(NewRayClusterSpec(MountConfigMap[rayv1ac.RayClusterSpecApplyConfiguration](jobs, "/home/ray/jobs")).
42+
WithNetworkPolicy(networkPolicy)).
43+
WithEntrypoint("python /home/ray/jobs/counter.py").
44+
WithRuntimeEnvYAML(`
45+
env_vars:
46+
counter_name: test_counter
47+
`).
48+
WithShutdownAfterJobFinishes(false).
49+
WithSubmitterPodTemplate(JobSubmitterPodTemplateApplyConfiguration()))
50+
51+
rayJob, err := test.Client().Ray().RayV1().RayJobs(namespace.Name).Apply(test.Ctx(), rayJobAC, TestApplyOptions)
52+
g.Expect(err).NotTo(HaveOccurred())
53+
LogWithTimestamp(test.T(), "Created RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)
54+
55+
// Wait for the operator to spin up the RayCluster.
56+
LogWithTimestamp(test.T(), "Waiting for RayJob %s/%s to start its RayCluster", rayJob.Namespace, rayJob.Name)
57+
g.Eventually(func(gg Gomega) string {
58+
rayJob, err = GetRayJob(test, rayJob.Namespace, rayJob.Name)
59+
gg.Expect(err).NotTo(HaveOccurred())
60+
return rayJob.Status.RayClusterName
61+
}, TestTimeoutShort).ShouldNot(BeEmpty())
62+
63+
rayClusterName := rayJob.Status.RayClusterName
64+
// Both policies should exist before the cluster can become ready.
65+
g.Eventually(func(gg Gomega) {
66+
_, err := test.Client().Core().NetworkingV1().NetworkPolicies(namespace.Name).
67+
Get(test.Ctx(), rayClusterName+"-head", metav1.GetOptions{})
68+
gg.Expect(err).NotTo(HaveOccurred())
69+
_, err = test.Client().Core().NetworkingV1().NetworkPolicies(namespace.Name).
70+
Get(test.Ctx(), rayClusterName+"-workers", metav1.GetOptions{})
71+
gg.Expect(err).NotTo(HaveOccurred())
72+
}, TestTimeoutShort).Should(Succeed())
73+
LogWithTimestamp(test.T(), "Head and worker NetworkPolicies created for RayCluster %s/%s", namespace.Name, rayClusterName)
74+
75+
// If the policies are right the cluster starts and the job completes.
76+
LogWithTimestamp(test.T(), "Waiting for RayJob %s/%s to complete", rayJob.Namespace, rayJob.Name)
77+
g.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutLong).
78+
Should(WithTransform(RayJobStatus, Satisfy(rayv1.IsJobTerminal)))
79+
80+
g.Expect(GetRayJob(test, rayJob.Namespace, rayJob.Name)).
81+
To(WithTransform(RayJobStatus, Equal(rayv1.JobStatusSucceeded)))
82+
}
83+
84+
// RayJob with clusterSelector pointing at a RayCluster we create ourselves, under
85+
// NetworkPolicy DenyAll. Since the RayCluster isn't owned by the RayJob, the
86+
// operator won't add the submitter ingress rule for us, so we add it by hand
87+
// (matching the submitter pod's labels) plus DNS egress.
88+
func TestRayJobWithClusterSelectorAndNetworkPolicy(t *testing.T) {
89+
test := With(t)
90+
g := NewWithT(t)
91+
92+
namespace := test.NewTestNamespace()
93+
94+
jobsAC := NewConfigMap(namespace.Name, Files(test, "counter.py"))
95+
jobs, err := test.Client().Core().CoreV1().ConfigMaps(namespace.Name).Apply(test.Ctx(), jobsAC, TestApplyOptions)
96+
g.Expect(err).NotTo(HaveOccurred())
97+
LogWithTimestamp(test.T(), "Created ConfigMap %s/%s successfully", jobs.Namespace, jobs.Name)
98+
99+
rayJobName := "counter"
100+
101+
// Head needs the submitter ingress rule (dashboard, default 8265) + DNS;
102+
// workers just need DNS.
103+
networkPolicy := rayv1ac.NetworkPolicyConfig().
104+
WithMode(rayv1.NetworkPolicyDenyAll).
105+
WithHead(rayv1ac.NetworkPolicyRules().
106+
WithIngressRules(SubmitterIngressRule(rayJobName, utils.DefaultDashboardPort)).
107+
WithEgressRules(DNSEgressRule())).
108+
WithWorker(rayv1ac.NetworkPolicyRules().WithEgressRules(DNSEgressRule()))
109+
110+
rayClusterAC := rayv1ac.RayCluster("raycluster", namespace.Name).
111+
WithSpec(NewRayClusterSpec(MountConfigMap[rayv1ac.RayClusterSpecApplyConfiguration](jobs, "/home/ray/jobs")).
112+
WithNetworkPolicy(networkPolicy))
113+
114+
rayCluster, err := test.Client().Ray().RayV1().RayClusters(namespace.Name).Apply(test.Ctx(), rayClusterAC, TestApplyOptions)
115+
g.Expect(err).NotTo(HaveOccurred())
116+
LogWithTimestamp(test.T(), "Created RayCluster %s/%s successfully", rayCluster.Namespace, rayCluster.Name)
117+
118+
// Sanity check that both policies landed.
119+
g.Eventually(func(gg Gomega) {
120+
_, err := test.Client().Core().NetworkingV1().NetworkPolicies(namespace.Name).
121+
Get(test.Ctx(), rayCluster.Name+"-head", metav1.GetOptions{})
122+
gg.Expect(err).NotTo(HaveOccurred())
123+
_, err = test.Client().Core().NetworkingV1().NetworkPolicies(namespace.Name).
124+
Get(test.Ctx(), rayCluster.Name+"-workers", metav1.GetOptions{})
125+
gg.Expect(err).NotTo(HaveOccurred())
126+
}, TestTimeoutShort).Should(Succeed())
127+
128+
LogWithTimestamp(test.T(), "Waiting for RayCluster %s/%s to become ready", rayCluster.Namespace, rayCluster.Name)
129+
g.Eventually(RayCluster(test, rayCluster.Namespace, rayCluster.Name), TestTimeoutLong).
130+
Should(WithTransform(RayClusterState, Equal(rayv1.Ready)))
131+
132+
// RayJob targeting the existing cluster by label.
133+
rayJobAC := rayv1ac.RayJob(rayJobName, namespace.Name).
134+
WithSpec(rayv1ac.RayJobSpec().
135+
WithClusterSelector(map[string]string{utils.RayClusterLabelKey: rayCluster.Name}).
136+
WithEntrypoint("python /home/ray/jobs/counter.py").
137+
WithRuntimeEnvYAML(`
138+
env_vars:
139+
counter_name: test_counter
140+
`).
141+
WithSubmitterPodTemplate(JobSubmitterPodTemplateApplyConfiguration()))
142+
143+
rayJob, err := test.Client().Ray().RayV1().RayJobs(namespace.Name).Apply(test.Ctx(), rayJobAC, TestApplyOptions)
144+
g.Expect(err).NotTo(HaveOccurred())
145+
LogWithTimestamp(test.T(), "Created RayJob %s/%s successfully", rayJob.Namespace, rayJob.Name)
146+
147+
// The submitter has to get through the head policy to reach the dashboard, so
148+
// the job succeeding is what proves our ingress rule is correct.
149+
LogWithTimestamp(test.T(), "Waiting for RayJob %s/%s to complete", rayJob.Namespace, rayJob.Name)
150+
g.Eventually(RayJob(test, rayJob.Namespace, rayJob.Name), TestTimeoutLong).
151+
Should(WithTransform(RayJobStatus, Satisfy(rayv1.IsJobTerminal)))
152+
153+
g.Expect(GetRayJob(test, rayJob.Namespace, rayJob.Name)).
154+
To(WithTransform(RayJobStatus, Equal(rayv1.JobStatusSucceeded)))
155+
}

‎ray-operator/test/e2erayservice/rayservice_network_policy_test.go‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,103 @@
1+
package e2erayservice
2+
3+
import (
4+
"testing"
5+
6+
. "github.com/onsi/gomega"
7+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
8+
9+
rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
10+
"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
11+
rayv1ac "github.com/ray-project/kuberay/ray-operator/pkg/client/applyconfiguration/ray/v1"
12+
. "github.com/ray-project/kuberay/ray-operator/test/support"
13+
)
14+
15+
// RayService under NetworkPolicy DenyAll. Checks the service comes up and, more
16+
// importantly, that we can actually hit the Serve app from an external pod. Under
17+
// DenyAll the operator only allows intra-cluster traffic, so we add DNS egress
18+
// (FQDN resolution), HTTPS egress (the sample pulls its working_dir from GitHub),
19+
// Serve-port ingress (so the curl pod can reach 8000), and operator ingress on
20+
// the dashboard (8265, push Serve config + poll status) and serving (8000, proxy
21+
// health probe) ports, since the reconciler dials the head directly on both and
22+
// without them the service never goes Ready. Kindnet enforces NetworkPolicy on
23+
// k8s >= 1.32 (CI runs on 1.35), so a wrong rule means the service never starts
24+
// or the curl gets blocked.
25+
func TestRayServiceWithNetworkPolicy(t *testing.T) {
26+
test := With(t)
27+
g := NewWithT(t)
28+
29+
namespace := test.NewTestNamespace()
30+
rayServiceName := "rayservice-network-policy"
31+
32+
networkPolicy := rayv1ac.NetworkPolicyConfig().
33+
WithMode(rayv1.NetworkPolicyDenyAll).
34+
WithHead(rayv1ac.NetworkPolicyRules().
35+
WithIngressRules(
36+
ServeIngressRule(utils.DefaultServingPort),
37+
// The reconciler hits the head directly on both the dashboard (push
38+
// Serve config / poll status) and the serving port (proxy health probe).
39+
OperatorIngressRule(utils.DefaultDashboardPort, utils.DefaultServingPort),
40+
).
41+
WithEgressRules(DNSEgressRule(), AllHostsHTTPSEgressRule())).
42+
WithWorker(rayv1ac.NetworkPolicyRules().
43+
WithIngressRules(ServeIngressRule(utils.DefaultServingPort)).
44+
WithEgressRules(DNSEgressRule(), AllHostsHTTPSEgressRule()))
45+
46+
rayServiceSpec := RayServiceSampleYamlApplyConfiguration()
47+
rayServiceSpec.RayClusterSpec.WithNetworkPolicy(networkPolicy)
48+
49+
rayServiceAC := rayv1ac.RayService(rayServiceName, namespace.Name).WithSpec(rayServiceSpec)
50+
51+
rayService, err := test.Client().Ray().RayV1().RayServices(namespace.Name).Apply(test.Ctx(), rayServiceAC, TestApplyOptions)
52+
g.Expect(err).NotTo(HaveOccurred())
53+
g.Expect(rayService).NotTo(BeNil())
54+
LogWithTimestamp(test.T(), "Created RayService %s/%s successfully with NetworkPolicy DenyAll", rayService.Namespace, rayService.Name)
55+
56+
defer func() {
57+
err := test.Client().Ray().RayV1().RayServices(namespace.Name).Delete(test.Ctx(), rayService.Name, metav1.DeleteOptions{})
58+
if err != nil {
59+
LogWithTimestamp(test.T(), "WARNING: Failed to delete RayService %s: %v", rayService.Name, err)
60+
}
61+
}()
62+
63+
LogWithTimestamp(test.T(), "Waiting for RayService %s/%s to be ready", rayService.Namespace, rayService.Name)
64+
g.Eventually(RayService(test, rayService.Namespace, rayService.Name), TestTimeoutLong).
65+
Should(WithTransform(IsRayServiceReady, BeTrue()))
66+
67+
rayService, err = GetRayService(test, namespace.Name, rayServiceName)
68+
g.Expect(err).NotTo(HaveOccurred())
69+
LogWithTimestamp(test.T(), "RayService %s/%s is ready", rayService.Namespace, rayService.Name)
70+
71+
rayClusterName := rayService.Status.ActiveServiceStatus.RayClusterName
72+
g.Expect(rayClusterName).NotTo(BeEmpty(), "RayCluster name should be populated in status")
73+
74+
// Sanity check that both policies landed.
75+
g.Eventually(func(gg Gomega) {
76+
_, err := test.Client().Core().NetworkingV1().NetworkPolicies(namespace.Name).
77+
Get(test.Ctx(), rayClusterName+"-head", metav1.GetOptions{})
78+
gg.Expect(err).NotTo(HaveOccurred())
79+
_, err = test.Client().Core().NetworkingV1().NetworkPolicies(namespace.Name).
80+
Get(test.Ctx(), rayClusterName+"-workers", metav1.GetOptions{})
81+
gg.Expect(err).NotTo(HaveOccurred())
82+
}, TestTimeoutShort).Should(Succeed())
83+
LogWithTimestamp(test.T(), "Head and worker NetworkPolicies created for RayCluster %s/%s", namespace.Name, rayClusterName)
84+
85+
// Spin up a curl pod outside the cluster and confirm it can actually reach
86+
// the Serve app through the ingress rule.
87+
curlPodName := "curl-pod"
88+
curlContainerName := "curl-container"
89+
LogWithTimestamp(test.T(), "Creating curl pod %s/%s", namespace.Name, curlPodName)
90+
curlPod, err := CreateCurlPod(g, test, curlPodName, curlContainerName, namespace.Name)
91+
g.Expect(err).NotTo(HaveOccurred())
92+
93+
LogWithTimestamp(test.T(), "Sending requests to the RayService Serve application")
94+
g.Eventually(func(gg Gomega) {
95+
stdout, _ := CurlRayServicePod(test, rayService, curlPod, curlContainerName, "/fruit", `["MANGO", 2]`)
96+
gg.Expect(stdout.String()).To(Equal("6"))
97+
}, TestTimeoutMedium).Should(Succeed())
98+
99+
stdout, _ := CurlRayServicePod(test, rayService, curlPod, curlContainerName, "/calc", `["MUL", 3]`)
100+
g.Expect(stdout.String()).To(Equal("15 pizzas please!"))
101+
102+
LogWithTimestamp(test.T(), "RayService %s/%s served requests successfully under NetworkPolicy DenyAll", rayService.Namespace, rayService.Name)
103+
}

0 commit comments

Comments
 (0)