Copilot commented on code in PR #105:
URL:
https://github.com/apache/cloudstack-kubernetes-provider/pull/105#discussion_r3902819809
##########
Makefile:
##########
@@ -53,6 +53,22 @@ ifneq (${GIT_IS_TAG},NOT_A_TAG)
docker tag apache/cloudstack-kubernetes-provider:${GIT_COMMIT_SHORT}
apache/cloudstack-kubernetes-provider:${GIT_TAG}
endif
+# Simulator-based e2e environment; see docs/development.md
+e2e-up:
+ hack/e2e/up.sh
+
+e2e-down:
+ hack/e2e/99-down.sh
+
+# go test runs with the package directory as its working directory, so
+# KUBECONFIG must be absolute.
+test-e2e:
+ @test -f hack/e2e/_out/keys.env || (echo "environment not up; run 'make
e2e-up' first" && exit 1)
+ . hack/e2e/_out/keys.env && \
+ KUBECONFIG=${CURDIR}/hack/e2e/_out/kubeconfig \
+ CS_API_URL=http://localhost:8080/client/api \
Review Comment:
`test-e2e` hardcodes `CS_API_URL` to port 8080, but the harness supports
overriding `SIM_HOST_PORT`/`CS_API_URL` via `hack/e2e/env.sh`. This makes `make
test-e2e` inconsistent with the harness and can fail for users/CI that override
ports. Consider honoring an existing `CS_API_URL` env var (only defaulting when
unset) or sourcing `hack/e2e/env.sh` to derive the correct endpoint.
##########
test/e2e/annotations_test.go:
##########
@@ -0,0 +1,176 @@
+//go:build e2e
+
+/*
+ * 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 e2e
+
+import (
+ "context"
+ "strings"
+ "testing"
+
+ "github.com/blang/semver/v4"
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+ annotationSourceCidrs =
"service.beta.kubernetes.io/cloudstack-load-balancer-source-cidrs"
+ annotationHostname =
"service.beta.kubernetes.io/cloudstack-load-balancer-hostname"
+ annotationIPAssociated =
"service.beta.kubernetes.io/cloudstack-load-balancer-ip-associated-by-controller"
//nolint:gosec
+)
+
+func TestAnnot_SourceCIDRs(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(func(s *corev1.Service) {
+ s.Annotations = map[string]string{
+ annotationSourceCidrs: "10.0.0.0/8,192.168.100.0/24",
+ }
+ })
+ lbName := defaultLoadBalancerName(svc)
+
+ f.WaitForIngressIP(svc)
+ rules := f.WaitForLBRules(lbName, 1)
+ for _, cidr := range []string{"10.0.0.0/8", "192.168.100.0/24"} {
+ if !strings.Contains(rules[0].Cidrlist, cidr) {
+ t.Errorf("rule cidrlist = %q, want it to contain %s",
rules[0].Cidrlist, cidr)
+ }
+ }
+ originalRuleID := rules[0].Id
+
+ // Change the CIDR list. On >= 4.22 the rule is updated in place (same
+ // ID); on older versions it is deleted and recreated (new ID).
+ f.UpdateService(svc, func(s *corev1.Service) {
+ s.Annotations[annotationSourceCidrs] = "172.16.0.0/12"
+ })
+ inPlace := f.Version.GTE(semver.Version{Major: 4, Minor: 22, Patch: 0})
+ f.Eventually(lbSyncTimeout, lbSyncInterval, "cidr list update to
propagate",
+ func() (bool, error) {
+ current, err := f.LBRules(lbName)
+ if err != nil || len(current) != 1 {
+ return false, err
+ }
+ if !strings.Contains(current[0].Cidrlist,
"172.16.0.0/12") {
+ return false, nil
+ }
+ if inPlace && current[0].Id != originalRuleID {
+ t.Errorf("expected in-place cidr update on %s
(rule ID changed %s -> %s)",
+ f.Version, originalRuleID,
current[0].Id)
+ }
+ if !inPlace && current[0].Id == originalRuleID {
+ t.Errorf("expected rule recreation on %s (rule
ID unchanged)", f.Version)
+ }
+ return true, nil
+ })
+}
+
+func TestAnnot_Hostname(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(func(s *corev1.Service) {
+ s.Annotations = map[string]string{
+ annotationHostname: "lb.example.com",
+ }
+ })
+
+ ingress := f.WaitForIngressIP(svc)
+ if ingress.Hostname != "lb.example.com" {
+ t.Errorf("ingress hostname = %q, want lb.example.com",
ingress.Hostname)
+ }
+ if ingress.IP != "" {
+ t.Errorf("ingress IP = %q, want empty when hostname annotation
is set", ingress.IP)
+ }
+}
+
+func TestAnnot_SessionAffinity(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(func(s *corev1.Service) {
+ s.Spec.SessionAffinity = corev1.ServiceAffinityClientIP
+ })
+ lbName := defaultLoadBalancerName(svc)
+
+ f.WaitForIngressIP(svc)
+ rules := f.WaitForLBRules(lbName, 1)
+ if rules[0].Algorithm != "source" {
+ t.Errorf("algorithm = %q, want source for sessionAffinity
ClientIP", rules[0].Algorithm)
+ }
+
+ f.UpdateService(svc, func(s *corev1.Service) {
+ s.Spec.SessionAffinity = corev1.ServiceAffinityNone
+ })
+ f.Eventually(lbSyncTimeout, lbSyncInterval, "algorithm to revert to
roundrobin",
+ func() (bool, error) {
+ current, err := f.LBRules(lbName)
+ if err != nil || len(current) != 1 {
+ return false, err
+ }
+ return current[0].Algorithm == "roundrobin", nil
+ })
+}
+
+func TestAnnot_ExplicitLoadBalancerIP(t *testing.T) {
+ f := NewFramework(t)
+
+ // Pick a free IP from the simulator's public range instead of
hardcoding
+ // one that a parallel test may have grabbed.
+ p := f.CS.Address.NewListPublicIpAddressesParams()
+ p.SetAllocatedonly(false)
+ p.SetListall(true)
+ p.SetState("Free")
+ resp, err := f.CS.Address.ListPublicIpAddresses(p)
Review Comment:
This test queries free public IPs without applying the framework’s project
scoping (`f.ProjectID`). Other framework helpers consistently set `projectid`
when running in project mode; doing the same here would make the test suite
more robust if it’s ever run with `CS_PROJECT_ID` set, and avoids accidentally
selecting an IP that’s not usable in the current scope. Consider `if
f.ProjectID != \"\" { p.SetProjectid(f.ProjectID) }` (or a small helper in
`Framework`).
##########
test/e2e/framework.go:
##########
@@ -0,0 +1,396 @@
+//go:build e2e
+
+/*
+ * 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 e2e contains end-to-end tests that run against a live Kubernetes
+// cluster whose cloud-controller-manager talks to a CloudStack management
+// server (normally the simulator brought up by hack/e2e/up.sh).
+//
+// Configuration comes from the environment:
+//
+// KUBECONFIG kubeconfig of the cluster under test
+// CS_API_URL CloudStack API endpoint (as reachable from the test
process)
+// CS_API_KEY CloudStack API key
+// CS_SECRET_KEY CloudStack secret key
+// CS_PROJECT_ID optional project scoping (set for the VPC phase)
+//
+// When any required variable is missing, the tests skip.
+package e2e
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "fmt"
+ "os"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/cloudstack-go/v2/cloudstack"
+ "github.com/blang/semver/v4"
+ corev1 "k8s.io/api/core/v1"
+ apierrors "k8s.io/apimachinery/pkg/api/errors"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/tools/clientcmd"
+)
+
+const (
+ lbSyncTimeout = 3 * time.Minute
+ lbSyncInterval = 3 * time.Second
+)
+
+// Framework bundles the clients and helpers shared by all e2e tests.
+type Framework struct {
+ T *testing.T
+ K8s kubernetes.Interface
+ CS *cloudstack.CloudStackClient
+ Namespace string
+ ProjectID string
+ Version semver.Version
+}
+
+// NewFramework builds clients from the environment, skipping the test when
+// the environment is not configured. It creates a per-test namespace that is
+// deleted on cleanup.
+func NewFramework(t *testing.T) *Framework {
+ t.Helper()
+
+ apiURL := os.Getenv("CS_API_URL")
+ apiKey := os.Getenv("CS_API_KEY")
+ secretKey := os.Getenv("CS_SECRET_KEY")
+ if apiURL == "" || apiKey == "" || secretKey == "" {
+ t.Skip("CS_API_URL/CS_API_KEY/CS_SECRET_KEY not set; skipping
e2e test")
+ }
+
+ kubeconfig := os.Getenv("KUBECONFIG")
+ if kubeconfig == "" {
+ t.Skip("KUBECONFIG not set; skipping e2e test")
+ }
+ restCfg, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
+ if err != nil {
+ t.Fatalf("building kubeconfig: %v", err)
+ }
+ k8s, err := kubernetes.NewForConfig(restCfg)
+ if err != nil {
+ t.Fatalf("building kubernetes client: %v", err)
+ }
+
+ verifySSL := true
+ if noVerify, err := strconv.ParseBool(os.Getenv("CS_SSL_NO_VERIFY"));
err == nil {
+ verifySSL = !noVerify
+ }
+ cs := cloudstack.NewAsyncClient(apiURL, apiKey, secretKey, verifySSL)
+
+ f := &Framework{
+ T: t,
+ K8s: k8s,
+ CS: cs,
+ ProjectID: os.Getenv("CS_PROJECT_ID"),
+ }
+ f.Version = f.managementServerVersion()
+ f.Namespace = f.createNamespace()
+ return f
+}
+
+func (f *Framework) managementServerVersion() semver.Version {
+ f.T.Helper()
+ resp, err := f.CS.Management.ListManagementServersMetrics(
+ f.CS.Management.NewListManagementServersMetricsParams())
+ if err != nil {
+ f.T.Fatalf("listing management servers: %v", err)
+ }
+ if resp.Count == 0 {
+ f.T.Fatal("no management servers found")
+ }
+ raw :=
strings.Join(strings.SplitN(resp.ManagementServersMetrics[0].Version, ".",
4)[0:3], ".")
Review Comment:
Slicing `[0:3]` can panic if the management server version string has fewer
than 3 dot-separated components. If CloudStack ever returns an unexpected
format (e.g., a shortened version string), this would crash the test binary
rather than failing/skipping cleanly. Consider extracting the first three
numeric components more defensively (check split length before slicing, or
parse tolerantly from the full string and then normalize).
##########
test/e2e/loadbalancer_test.go:
##########
@@ -0,0 +1,213 @@
+//go:build e2e
+
+/*
+ * 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 e2e
+
+import (
+ "context"
+ "fmt"
+ "net"
+ "strconv"
+ "strings"
+ "testing"
+
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestLB_CreateSingleTCPPort(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(nil)
+ lbName := defaultLoadBalancerName(svc)
+
+ ingress := f.WaitForIngressIP(svc)
+ if ingress.IP == "" {
+ t.Fatalf("expected an ingress IP, got %+v", ingress)
+ }
+ if ip := net.ParseIP(ingress.IP); ip == nil {
+ t.Fatalf("ingress IP %q is not a valid IP", ingress.IP)
+ }
+
+ rules := f.WaitForLBRules(lbName, 1)
+ rule := rules[0]
+ wantName := fmt.Sprintf("%s-tcp-80", lbName)
+ if rule.Name != wantName {
+ t.Errorf("rule name = %q, want %q", rule.Name, wantName)
+ }
+ if rule.Algorithm != "roundrobin" {
+ t.Errorf("rule algorithm = %q, want roundrobin", rule.Algorithm)
+ }
+ if rule.Publicport != "80" {
+ t.Errorf("rule public port = %q, want 80", rule.Publicport)
+ }
+ // The private port must be the service's NodePort.
+ current, err :=
f.K8s.CoreV1().Services(svc.Namespace).Get(context.Background(), svc.Name,
metav1.GetOptions{})
+ if err != nil {
+ t.Fatalf("getting service: %v", err)
+ }
+ nodePort := strconv.Itoa(int(current.Spec.Ports[0].NodePort))
+ if rule.Privateport != nodePort {
+ t.Errorf("rule private port = %q, want NodePort %q",
rule.Privateport, nodePort)
+ }
+ if rule.Publicip != ingress.IP {
+ t.Errorf("rule public IP = %q, want ingress IP %q",
rule.Publicip, ingress.IP)
+ }
+ if !strings.Contains(rule.Cidrlist, "0.0.0.0/0") {
+ t.Errorf("rule cidrlist = %q, want it to contain 0.0.0.0/0",
rule.Cidrlist)
+ }
+
+ // The isolated network offering supports the Firewall service, so a
+ // firewall rule must exist for the port.
+ f.Eventually(lbSyncTimeout, lbSyncInterval, "firewall rule for port 80",
+ func() (bool, error) {
+ fwRules, err := f.FirewallRules(rule.Publicipid)
+ if err != nil {
+ return false, err
+ }
+ for _, fw := range fwRules {
+ if fw.Startport == 80 && fw.Endport == 80 &&
strings.EqualFold(fw.Protocol, "tcp") {
+ return true, nil
+ }
+ }
+ return false, nil
+ })
+}
+
+func TestLB_MultiPort(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(func(s *corev1.Service) {
+ s.Spec.Ports = []corev1.ServicePort{
+ {Name: "http", Port: 80, Protocol: corev1.ProtocolTCP},
+ {Name: "https", Port: 443, Protocol:
corev1.ProtocolTCP},
+ }
+ })
+ lbName := defaultLoadBalancerName(svc)
+
+ f.WaitForIngressIP(svc)
+ rules := f.WaitForLBRules(lbName, 2)
+ if rules[0].Publicipid != rules[1].Publicipid {
+ t.Errorf("expected both rules to share a public IP, got %q and
%q",
+ rules[0].Publicipid, rules[1].Publicipid)
+ }
+ ports := map[string]bool{}
+ for _, r := range rules {
+ ports[r.Publicport] = true
+ }
+ if !ports["80"] || !ports["443"] {
+ t.Errorf("expected rules for ports 80 and 443, got %v", ports)
+ }
+}
+
+func TestLB_NodeMembership(t *testing.T) {
+ f := NewFramework(t)
+ svc := f.CreateLBService(nil)
+ lbName := defaultLoadBalancerName(svc)
+
+ f.WaitForIngressIP(svc)
+ rules := f.WaitForLBRules(lbName, 1)
+
+ // Only schedulable workers participate; kubeadm labels the control
plane
+ // node.kubernetes.io/exclude-from-external-load-balancers.
+ wantIDs := map[string]bool{}
+ for _, node := range f.Nodes() {
+ if _, excluded :=
node.Labels["node.kubernetes.io/exclude-from-external-load-balancers"];
excluded {
+ continue
+ }
+ vm, err := f.VMByName(node.Name)
+ if err != nil || vm == nil {
+ t.Fatalf("looking up VM for node %s: %v", node.Name,
err)
+ }
+ wantIDs[vm.Id] = true
+ }
+ if len(wantIDs) == 0 {
+ t.Fatal("no candidate worker nodes found")
+ }
+
+ f.Eventually(lbSyncTimeout, lbSyncInterval, "load balancer rule
instances to match worker VMs",
+ func() (bool, error) {
+ p :=
f.CS.LoadBalancer.NewListLoadBalancerRuleInstancesParams(rules[0].Id)
Review Comment:
This CloudStack query bypasses the framework’s project scoping. For
consistency with `Framework` helpers (and to reduce surprises if these tests
are run in project mode), consider setting `p.SetProjectid(f.ProjectID)` when
`f.ProjectID` is non-empty, or routing the call through a framework method that
applies scoping uniformly.
##########
test/e2e/framework.go:
##########
@@ -0,0 +1,396 @@
+//go:build e2e
+
+/*
+ * 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 e2e contains end-to-end tests that run against a live Kubernetes
+// cluster whose cloud-controller-manager talks to a CloudStack management
+// server (normally the simulator brought up by hack/e2e/up.sh).
+//
+// Configuration comes from the environment:
+//
+// KUBECONFIG kubeconfig of the cluster under test
+// CS_API_URL CloudStack API endpoint (as reachable from the test
process)
+// CS_API_KEY CloudStack API key
+// CS_SECRET_KEY CloudStack secret key
+// CS_PROJECT_ID optional project scoping (set for the VPC phase)
+//
+// When any required variable is missing, the tests skip.
+package e2e
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "fmt"
+ "os"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/cloudstack-go/v2/cloudstack"
+ "github.com/blang/semver/v4"
+ corev1 "k8s.io/api/core/v1"
+ apierrors "k8s.io/apimachinery/pkg/api/errors"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/tools/clientcmd"
+)
+
+const (
+ lbSyncTimeout = 3 * time.Minute
+ lbSyncInterval = 3 * time.Second
+)
+
+// Framework bundles the clients and helpers shared by all e2e tests.
+type Framework struct {
+ T *testing.T
+ K8s kubernetes.Interface
+ CS *cloudstack.CloudStackClient
+ Namespace string
+ ProjectID string
+ Version semver.Version
+}
+
+// NewFramework builds clients from the environment, skipping the test when
+// the environment is not configured. It creates a per-test namespace that is
+// deleted on cleanup.
+func NewFramework(t *testing.T) *Framework {
+ t.Helper()
+
+ apiURL := os.Getenv("CS_API_URL")
+ apiKey := os.Getenv("CS_API_KEY")
+ secretKey := os.Getenv("CS_SECRET_KEY")
+ if apiURL == "" || apiKey == "" || secretKey == "" {
+ t.Skip("CS_API_URL/CS_API_KEY/CS_SECRET_KEY not set; skipping
e2e test")
+ }
+
+ kubeconfig := os.Getenv("KUBECONFIG")
+ if kubeconfig == "" {
+ t.Skip("KUBECONFIG not set; skipping e2e test")
+ }
+ restCfg, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
+ if err != nil {
+ t.Fatalf("building kubeconfig: %v", err)
+ }
+ k8s, err := kubernetes.NewForConfig(restCfg)
+ if err != nil {
+ t.Fatalf("building kubernetes client: %v", err)
+ }
+
+ verifySSL := true
+ if noVerify, err := strconv.ParseBool(os.Getenv("CS_SSL_NO_VERIFY"));
err == nil {
+ verifySSL = !noVerify
+ }
+ cs := cloudstack.NewAsyncClient(apiURL, apiKey, secretKey, verifySSL)
+
+ f := &Framework{
+ T: t,
+ K8s: k8s,
+ CS: cs,
+ ProjectID: os.Getenv("CS_PROJECT_ID"),
+ }
+ f.Version = f.managementServerVersion()
+ f.Namespace = f.createNamespace()
+ return f
+}
+
+func (f *Framework) managementServerVersion() semver.Version {
+ f.T.Helper()
+ resp, err := f.CS.Management.ListManagementServersMetrics(
+ f.CS.Management.NewListManagementServersMetricsParams())
+ if err != nil {
+ f.T.Fatalf("listing management servers: %v", err)
+ }
+ if resp.Count == 0 {
+ f.T.Fatal("no management servers found")
+ }
+ raw :=
strings.Join(strings.SplitN(resp.ManagementServersMetrics[0].Version, ".",
4)[0:3], ".")
+ v, err := semver.ParseTolerant(raw)
+ if err != nil {
+ f.T.Fatalf("parsing management server version %q: %v", raw, err)
+ }
+ return v
+}
+
+func (f *Framework) createNamespace() string {
+ f.T.Helper()
+ buf := make([]byte, 4)
+ if _, err := rand.Read(buf); err != nil {
+ f.T.Fatalf("generating namespace suffix: %v", err)
+ }
+ name := "ccm-e2e-" + hex.EncodeToString(buf)
+ _, err := f.K8s.CoreV1().Namespaces().Create(context.Background(),
+ &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: name}},
metav1.CreateOptions{})
+ if err != nil {
+ f.T.Fatalf("creating namespace %s: %v", name, err)
+ }
+ f.T.Cleanup(func() {
+ _ = f.K8s.CoreV1().Namespaces().Delete(context.Background(),
name, metav1.DeleteOptions{})
+ })
+ return name
+}
+
+// Eventually polls cond until it returns true or the timeout elapses.
+func (f *Framework) Eventually(timeout, interval time.Duration, desc string,
cond func() (bool, error)) {
+ f.T.Helper()
+ deadline := time.Now().Add(timeout)
+ var lastErr error
+ for time.Now().Before(deadline) {
+ ok, err := cond()
+ lastErr = err
+ if ok {
+ return
+ }
+ time.Sleep(interval)
+ }
+ f.T.Fatalf("timed out after %s waiting for %s (last error: %v)",
timeout, desc, lastErr)
+}
+
+// CreateLBService creates a LoadBalancer service in the test namespace and
+// registers cleanup that deletes it and waits for the CloudStack rules to
+// disappear, failing the test if they do not. Later tests share this
+// simulator and its public IP pool, so a leaked rule has to be reported here
+// rather than left to surface as an unrelated failure downstream.
+func (f *Framework) CreateLBService(mutate func(*corev1.Service))
*corev1.Service {
+ f.T.Helper()
+ svc := &corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "e2e",
+ Namespace: f.Namespace,
+ },
+ Spec: corev1.ServiceSpec{
+ Type: corev1.ServiceTypeLoadBalancer,
+ Selector: map[string]string{"app": "e2e"},
+ Ports: []corev1.ServicePort{
+ {Name: "http", Port: 80, Protocol:
corev1.ProtocolTCP},
+ },
+ },
+ }
+ if mutate != nil {
+ mutate(svc)
+ }
+ created, err := f.K8s.CoreV1().Services(f.Namespace).Create(
+ context.Background(), svc, metav1.CreateOptions{})
+ if err != nil {
+ f.T.Fatalf("creating service: %v", err)
+ }
+ f.T.Cleanup(func() { f.DeleteServiceAndWait(created) })
+ return created
+}
+
+// DeleteServiceAndWait deletes the service (if it still exists) and waits for
+// its CloudStack load balancer rules to be cleaned up.
+func (f *Framework) DeleteServiceAndWait(svc *corev1.Service) {
+ f.T.Helper()
+ err := f.K8s.CoreV1().Services(svc.Namespace).Delete(
+ context.Background(), svc.Name, metav1.DeleteOptions{})
+ if apierrors.IsNotFound(err) {
+ return
+ }
+ if err != nil {
+ // Anything other than NotFound means the service may still
exist, so
+ // skipping the cleanup wait below would leak CloudStack rules
into
+ // later tests sharing this simulator.
+ f.T.Fatalf("deleting service %s/%s: %v", svc.Namespace,
svc.Name, err)
+ }
+ lbName := defaultLoadBalancerName(svc)
+ deadline := time.Now().Add(lbSyncTimeout)
+ var lastErr error
+ for time.Now().Before(deadline) {
+ rules, err := f.LBRules(lbName)
+ lastErr = err
+ if err == nil && len(rules) == 0 {
+ return
+ }
+ time.Sleep(lbSyncInterval)
+ }
+ // Errorf rather than Fatalf: this usually runs from t.Cleanup, and the
+ // remaining cleanups still need to run. lastErr distinguishes "the
rules
+ // are still there" from "listing them kept failing".
+ f.T.Errorf("load balancer rules for %s were not cleaned up within %s "+
+ "(last list error: %v)", lbName, lbSyncTimeout, lastErr)
+}
+
+// defaultLoadBalancerName mirrors cloudprovider.DefaultLoadBalancerName: "a"
+// followed by the service UID with dashes stripped, truncated to 32 chars.
+func defaultLoadBalancerName(svc *corev1.Service) string {
+ name := "a" + strings.ReplaceAll(string(svc.UID), "-", "")
+ if len(name) > 32 {
+ name = name[:32]
+ }
+ return name
+}
+
+// LBRules returns the CloudStack load balancer rules whose names start with
+// the given LB name.
+func (f *Framework) LBRules(lbName string) ([]*cloudstack.LoadBalancerRule,
error) {
+ p := f.CS.LoadBalancer.NewListLoadBalancerRulesParams()
+ p.SetKeyword(lbName)
+ p.SetListall(true)
+ if f.ProjectID != "" {
+ p.SetProjectid(f.ProjectID)
+ }
+ resp, err := f.CS.LoadBalancer.ListLoadBalancerRules(p)
+ if err != nil {
+ return nil, err
+ }
+ var rules []*cloudstack.LoadBalancerRule
+ for _, r := range resp.LoadBalancerRules {
+ if strings.HasPrefix(r.Name, lbName) {
+ rules = append(rules, r)
+ }
+ }
+ return rules, nil
+}
+
+// WaitForIngressIP waits until the service has a load balancer ingress entry
+// and returns it.
+func (f *Framework) WaitForIngressIP(svc *corev1.Service)
corev1.LoadBalancerIngress {
+ f.T.Helper()
+ var ingress corev1.LoadBalancerIngress
+ f.Eventually(lbSyncTimeout, lbSyncInterval,
+ fmt.Sprintf("service %s/%s to get an ingress address",
svc.Namespace, svc.Name),
+ func() (bool, error) {
+ current, err :=
f.K8s.CoreV1().Services(svc.Namespace).Get(
+ context.Background(), svc.Name,
metav1.GetOptions{})
+ if err != nil {
+ return false, err
+ }
+ if len(current.Status.LoadBalancer.Ingress) == 0 {
+ return false, nil
+ }
+ ingress = current.Status.LoadBalancer.Ingress[0]
+ return true, nil
+ })
+ return ingress
+}
+
+// WaitForLBRules waits until exactly want rules exist for lbName and returns
them.
+func (f *Framework) WaitForLBRules(lbName string, want int)
[]*cloudstack.LoadBalancerRule {
+ f.T.Helper()
+ var rules []*cloudstack.LoadBalancerRule
+ f.Eventually(lbSyncTimeout, lbSyncInterval,
+ fmt.Sprintf("%d load balancer rule(s) named %s-*", want,
lbName),
+ func() (bool, error) {
+ var err error
+ rules, err = f.LBRules(lbName)
+ if err != nil {
+ return false, err
+ }
+ return len(rules) == want, nil
+ })
+ return rules
+}
+
+// FirewallRules lists the firewall rules on a public IP.
+func (f *Framework) FirewallRules(publicIPID string)
([]*cloudstack.FirewallRule, error) {
+ p := f.CS.Firewall.NewListFirewallRulesParams()
+ p.SetIpaddressid(publicIPID)
+ p.SetListall(true)
+ if f.ProjectID != "" {
+ p.SetProjectid(f.ProjectID)
+ }
+ resp, err := f.CS.Firewall.ListFirewallRules(p)
+ if err != nil {
+ return nil, err
+ }
+ return resp.FirewallRules, nil
+}
+
+// ACLRules lists the network ACL rules on an ACL list.
+func (f *Framework) ACLRules(aclListID string) ([]*cloudstack.NetworkACL,
error) {
+ p := f.CS.NetworkACL.NewListNetworkACLsParams()
+ p.SetAclid(aclListID)
+ p.SetListall(true)
+ if f.ProjectID != "" {
+ p.SetProjectid(f.ProjectID)
+ }
+ resp, err := f.CS.NetworkACL.ListNetworkACLs(p)
+ if err != nil {
+ return nil, err
+ }
+ return resp.NetworkACLs, nil
+}
+
+// PublicIP fetches a public IP address record by its ID.
+func (f *Framework) PublicIP(id string) (*cloudstack.PublicIpAddress, error) {
+ p := f.CS.Address.NewListPublicIpAddressesParams()
+ p.SetId(id)
+ p.SetListall(true)
+ p.SetAllocatedonly(false)
+ if f.ProjectID != "" {
+ p.SetProjectid(f.ProjectID)
+ }
+ resp, err := f.CS.Address.ListPublicIpAddresses(p)
+ if err != nil {
+ return nil, err
+ }
+ if resp.Count == 0 {
+ return nil, nil
+ }
+ return resp.PublicIpAddresses[0], nil
+}
+
+// VMByName returns the CloudStack VM with the given name, or nil.
+func (f *Framework) VMByName(name string) (*cloudstack.VirtualMachine, error) {
+ vm, count, err := f.CS.VirtualMachine.GetVirtualMachineByName(
+ name, cloudstack.WithProject(f.ProjectID))
+ if err != nil {
+ if count == 0 {
+ return nil, nil
+ }
+ return nil, err
+ }
+ return vm, nil
+}
+
+// Nodes returns all nodes of the cluster under test.
+func (f *Framework) Nodes() []corev1.Node {
+ f.T.Helper()
+ nodes, err := f.K8s.CoreV1().Nodes().List(context.Background(),
metav1.ListOptions{})
+ if err != nil {
+ f.T.Fatalf("listing nodes: %v", err)
+ }
+ return nodes.Items
+}
+
+// UpdateService applies mutate to the latest version of the service and
+// updates it, retrying on conflicts.
+func (f *Framework) UpdateService(svc *corev1.Service, mutate
func(*corev1.Service)) *corev1.Service {
Review Comment:
The comment says this retries on conflicts specifically, but the
implementation retries on any update error (it doesn’t distinguish conflict
errors from other failures). Either update the docstring to match the behavior,
or explicitly detect conflict errors (retry) while returning non-conflict
errors immediately to fail faster and make debugging easier.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
