add service connectivity test

test/framework/resources/k8s/manifest/service.go

@@ -14,6 +14,8 @@
 package manifest
 
 import (
+	"github.com/aws/amazon-vpc-cni-k8s/test/framework/utils"
+
 	v1 "k8s.io/api/core/v1"
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -26,14 +28,17 @@ type ServiceBuilder struct {
 	nodePort    int32
 	protocol    v1.Protocol
 	selector    map[string]string
+	annotation  map[string]string
 	serviceType v1.ServiceType
 }
 
 func NewHTTPService() *ServiceBuilder {
 	return &ServiceBuilder{
-		port:     80,
-		protocol: v1.ProtocolTCP,
-		selector: map[string]string{},
+		namespace:  utils.DefaultTestNamespace,
+		port:       80,
+		protocol:   v1.ProtocolTCP,
+		selector:   map[string]string{},
+		annotation: map[string]string{},
 	}
 }
 
@@ -72,11 +77,17 @@ func (s *ServiceBuilder) ServiceType(serviceType v1.ServiceType) *ServiceBuilder
 	return s
 }
 
-func (s *ServiceBuilder) Build() v1.Service {
-	return v1.Service{
+func (s *ServiceBuilder) Annotations(annotations map[string]string) *ServiceBuilder {
+	s.annotation = annotations
+	return s
+}
+
+func (s *ServiceBuilder) Build() *v1.Service {
+	return &v1.Service{
 		ObjectMeta: metaV1.ObjectMeta{
-			Name:      s.name,
-			Namespace: s.namespace,
+			Name:        s.name,
+			Namespace:   s.namespace,
+			Annotations: s.annotation,
 		},
 		Spec: v1.ServiceSpec{
 			Ports: []v1.ServicePort{{

test/framework/resources/k8s/resources/job.go

@@ -21,12 +21,14 @@ import (
 	"github.com/aws/amazon-vpc-cni-k8s/test/framework/utils"
 
 	v1 "k8s.io/api/batch/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type JobManager interface {
 	CreateAndWaitTillJobCompleted(job *v1.Job) (*v1.Job, error)
+	DeleteAndWaitTillJobIsDeleted(job *v1.Job) error
 }
 
 type defaultJobManager struct {
@@ -59,3 +61,21 @@ func (d *defaultJobManager) CreateAndWaitTillJobCompleted(job *v1.Job) (*v1.Job,
 		return false, nil
 	}, ctx.Done())
 }
+
+func (d *defaultJobManager) DeleteAndWaitTillJobIsDeleted(job *v1.Job) error {
+	ctx := context.Background()
+	err := d.k8sClient.Delete(ctx, job)
+	if err != nil {
+		return err
+	}
+	observedJob := &v1.Job{}
+	return wait.PollImmediateUntil(utils.PollIntervalShort, func() (bool, error) {
+		if err := d.k8sClient.Get(ctx, utils.NamespacedName(job), observedJob); err != nil {
+			if errors.IsNotFound(err) {
+				return true, nil
+			}
+			return false, err
+		}
+		return false, nil
+	}, ctx.Done())
+}

test/framework/utils/const.go

@@ -20,5 +20,7 @@ const (
 )
 
 const (
-	PollIntervalShort = time.Second * 2
+	PollIntervalShort  = time.Second * 2
+	PollIntervalMedium = time.Second * 5
+	PollIntervalLong   = time.Second * 20
 )

test/integration-new/cni/service_connectivity_test.go

@@ -0,0 +1,177 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may
+// not use this file except in compliance with the License. A copy of the
+// License is located at
+//
+//     http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file 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 cni
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/aws/amazon-vpc-cni-k8s/test/framework/resources/k8s/manifest"
+	"github.com/aws/amazon-vpc-cni-k8s/test/framework/utils"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	appsV1 "k8s.io/api/apps/v1"
+	batchV1 "k8s.io/api/batch/v1"
+	v1 "k8s.io/api/core/v1"
+)
+
+const (
+	serviceLabelSelectorKey = "role"
+	serviceLabelSelectorVal = "service-test"
+)
+
+// Verifies connectivity to deployment behind different service types
+var _ = Describe("test service connectivity", func() {
+	var err error
+
+	// Deployment running the http server
+	var deployment *appsV1.Deployment
+	var deploymentContainer v1.Container
+
+	// Service front ending the http server deployment
+	var service *v1.Service
+	var serviceType v1.ServiceType
+	var serviceAnnotation map[string]string
+
+	// Test job that verifies connectivity to the http server
+	// by querying the service URL
+	var testerJob *batchV1.Job
+	var testerContainer v1.Container
+
+	// Test job that verifies test fails when connecting to
+	// a non reachable port/address
+	var negativeTesterJob *batchV1.Job
+	var negativeTesterContainer v1.Container
+
+	JustBeforeEach(func() {
+		deploymentContainer = manifest.NewBusyBoxContainerBuilder().
+			Image("python").
+			Command([]string{"python3"}).
+			Args([]string{"-m", "http.server", "80"}).Build()
+
+		deployment = manifest.NewDefaultDeploymentBuilder().
+			Name("http-server").
+			Container(deploymentContainer).
+			Replicas(20).
+			PodLabel(serviceLabelSelectorKey, serviceLabelSelectorVal).
+			Build()
+
+		By("creating and waiting for deployment to be ready")
+		deployment, err = f.K8sResourceManagers.DeploymentManager().
+			CreateAndWaitTillDeploymentIsReady(deployment)
+		Expect(err).ToNot(HaveOccurred())
+
+		service = manifest.NewHTTPService().
+			ServiceType(serviceType).
+			Name("test-service").
+			Selector(serviceLabelSelectorKey, serviceLabelSelectorVal).
+			Annotations(serviceAnnotation).
+			Build()
+
+		By(fmt.Sprintf("creating the service of type %s", serviceType))
+		service, err = f.K8sResourceManagers.ServiceManager().
+			CreateService(context.Background(), service)
+		Expect(err).ToNot(HaveOccurred())
+
+		fmt.Fprintf(GinkgoWriter, "created service\n: %+v\n", service.Status)
+
+		By("sleeping for some time to allow service to become ready")
+		time.Sleep(utils.PollIntervalLong)
+
+		testerContainer = manifest.NewBusyBoxContainerBuilder().
+			Command([]string{"wget"}).
+			Args([]string{"--spider", "-T", "1", fmt.Sprintf("%s:%d", service.Spec.ClusterIP,
+				service.Spec.Ports[0].Port)}).
+			Build()
+
+		testerJob = manifest.NewDefaultJobBuilder().
+			Parallelism(20).
+			Container(testerContainer).
+			Build()
+
+		By("creating jobs to verify service connectivity")
+		_, err = f.K8sResourceManagers.JobManager().
+			CreateAndWaitTillJobCompleted(testerJob)
+		Expect(err).ToNot(HaveOccurred())
+
+		// Test connection to an unreachable port should fail
+		negativeTesterContainer = manifest.NewBusyBoxContainerBuilder().
+			Command([]string{"wget"}).
+			Args([]string{"--spider", "-T", "1", fmt.Sprintf("%s:%d", service.Spec.ClusterIP, 2273)}).
+			Build()
+
+		negativeTesterJob = manifest.NewDefaultJobBuilder().
+			Name("negative-test-job").
+			Parallelism(2).
+			Container(negativeTesterContainer).
+			Build()
+
+		By("creating negative jobs to verify service connectivity fails for unreachable port")
+		_, err = f.K8sResourceManagers.JobManager().
+			CreateAndWaitTillJobCompleted(negativeTesterJob)
+		Expect(err).To(HaveOccurred())
+	})
+
+	JustAfterEach(func() {
+		err := f.K8sResourceManagers.JobManager().DeleteAndWaitTillJobIsDeleted(testerJob)
+		Expect(err).ToNot(HaveOccurred())
+
+		err = f.K8sResourceManagers.JobManager().DeleteAndWaitTillJobIsDeleted(negativeTesterJob)
+		Expect(err).ToNot(HaveOccurred())
+
+		err = f.K8sResourceManagers.ServiceManager().DeleteService(context.Background(), service)
+		Expect(err).ToNot(HaveOccurred())
+
+		err = f.K8sResourceManagers.DeploymentManager().DeleteAndWaitTillDeploymentIsDeleted(deployment)
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	Context("when a deployment behind clb service is created", func() {
+		BeforeEach(func() {
+			serviceType = v1.ServiceTypeLoadBalancer
+		})
+
+		It("clb service pod should be reachable", func() {})
+	})
+
+	Context("when a deployment behind nlb service is created", func() {
+		BeforeEach(func() {
+			serviceType = v1.ServiceTypeLoadBalancer
+			serviceAnnotation = map[string]string{"service.beta.kubernetes.io/" +
+				"aws-load-balancer-type": "nlb"}
+		})
+
+		It("nlb service pod should be reachable", func() {})
+	})
+
+	Context("when a deployment behind cluster IP is created", func() {
+		BeforeEach(func() {
+			serviceType = v1.ServiceTypeClusterIP
+		})
+
+		It("clusterIP service pod should be reachable", func() {})
+	})
+
+	Context("when a deployment behind node port is created", func() {
+		BeforeEach(func() {
+			serviceType = v1.ServiceTypeNodePort
+		})
+
+		It("node port service pod should be reachable", func() {})
+	})
+
+	//TODO: Add test case to install lb controller and test with nlb-ip mode
+})