Add a external link case to integration test

integration/bosh_links_test.go

@@ -3,7 +3,6 @@ package integration_test
 import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-
 	corev1 "k8s.io/api/core/v1"
 
 	bm "code.cloudfoundry.org/cf-operator/testing/boshmanifest"
@@ -68,4 +67,37 @@ var _ = Describe("BOSHLinks", func() {
 			})
 		})
 	})
+
+	Context("when deployment has explicit external link dependencies", func() {
+		BeforeEach(func() {
+			natsConfigMap := env.Catalog.NatsConfigMap(deploymentName)
+			tearDown, err := env.CreateConfigMap(env.Namespace, natsConfigMap)
+			Expect(err).NotTo(HaveOccurred())
+			tearDowns = append(tearDowns, tearDown)
+
+			natsSecret := env.Catalog.NatsSecret(deploymentName)
+			tearDown, err = env.CreateSecret(env.Namespace, natsSecret)
+			Expect(err).NotTo(HaveOccurred())
+			tearDowns = append(tearDowns, tearDown)
+
+			natsPod := env.Catalog.NatsPod(deploymentName)
+			tearDown, err = env.CreatePod(env.Namespace, natsPod)
+			Expect(err).NotTo(HaveOccurred())
+			tearDowns = append(tearDowns, tearDown)
+
+			natsService := env.Catalog.NatsService(deploymentName)
+			tearDown, err = env.CreateService(env.Namespace, natsService)
+			Expect(err).NotTo(HaveOccurred())
+			tearDowns = append(tearDowns, tearDown)
+
+			boshManifest = env.BOSHManifestSecret(manifestRef, bm.NatsSmokeTestWithExternalLinks)
+		})
+
+		It("creates a secret for each link", func() {
+			By("waiting for job rendering done", func() {
+				err := env.WaitForPods(env.Namespace,"quarks.cloudfoundry.org/instance-group-name=nats-smoke-tests")
+				Expect(err).NotTo(HaveOccurred())
+			})
+		})
+	})
 })

pkg/bosh/manifest/cmd_instance_group_resolver.go

@@ -216,6 +216,8 @@ func (igr *InstanceGroupResolver) CollectQuarksLinks(linksPath string) error {
 
 			linkType := q.Type
 			properties := map[string]interface{}{}
+			properties[linkName] = map[string]interface{}{}
+			linkP := map[string]interface{}{}
 			err = afero.Walk(igr.fs, filepath.Clean(linksPath+"/"+l.Name()), func(path string, info os.FileInfo, err error) error {
 				if err != nil {
 					return err
@@ -225,22 +227,22 @@ func (igr *InstanceGroupResolver) CollectQuarksLinks(linksPath string) error {
 					_, propertyFileName := filepath.Split(path)
 					// Skip the symlink to a directory
 					if strings.HasPrefix(propertyFileName, "..") {
-						return filepath.SkipDir
+						return nil
 					}
 					varBytes, err := afero.ReadFile(igr.fs, path)
 					if err != nil {
 						return errors.Wrapf(err, "could not read link %s", l.Name())
 					}
 
-					properties[propertyFileName] = string(varBytes)
-
+					linkP[propertyFileName] = string(varBytes)
 				}
 				return nil
 			})
 			if err != nil {
 				return errors.Wrapf(err, "Walking links path")
 			}
 
+			properties[linkName] = linkP
 			err = igr.jobProviderLinks.AddExternalLink(linkName, linkType, q.Address, q.Instances, properties)
 			if err != nil {
 				return errors.Wrapf(err, "Collecting external link failed for %s", linkName)

pkg/bosh/manifest/cmd_instance_group_resolver_test.go

@@ -242,8 +242,8 @@ var _ = Describe("InstanceGroupResolver", func() {
 						expectedProperties := JobLinkProperties{
 							"doppler": map[string]interface{}{
 								"grpc_port": json.Number("7765"),
+								"fooprop":   json.Number("10001"),
 							},
-							"fooprop": json.Number("10001"),
 						}
 
 						Expect(deep.Equal(jobConsumesFromDoppler.Properties, expectedProperties)).To(HaveLen(0))
@@ -320,16 +320,16 @@ var _ = Describe("InstanceGroupResolver", func() {
 					_, err = fileP1.WriteString("fake_prop")
 					Expect(err).NotTo(HaveOccurred())
 
-					fileP2, err := fs.Create(converter.VolumeLinksPath + "doppler/doppler")
+					fileP2, err := fs.Create(converter.VolumeLinksPath + "doppler/grpc_port")
 					Expect(err).NotTo(HaveOccurred())
 					defer fileP2.Close()
-					_, err = fileP2.WriteString(`grpc_port: 7765`)
+					_, err = fileP2.WriteString(`7765`)
 					Expect(err).NotTo(HaveOccurred())
 
 				})
 
 				It("stores all the links of the instance group in a file", func() {
-					err = igr.CollectQuarksLinks("/var/run/secrets/links/")
+					err = igr.CollectQuarksLinks(converter.VolumeLinksPath)
 					Expect(err).ToNot(HaveOccurred())
 
 					m, err := igr.Manifest(true)
@@ -348,8 +348,10 @@ var _ = Describe("InstanceGroupResolver", func() {
 							},
 						},
 						Properties: JobLinkProperties{
-							"doppler": "grpc_port: 7765",
-							"fooprop": "fake_prop",
+							"doppler": map[string]interface{}{
+								"grpc_port": "7765",
+								"fooprop":   "fake_prop",
+							},
 						},
 					}))
 

pkg/bosh/manifest/containerization.go

@@ -81,7 +81,7 @@ type PostStartCondition struct {
 	Exec *corev1.ExecAction `json:"exec,omitempty"`
 }
 
-// QuarksLinks represents the links to share/discover information between BOSH and Kube Native components
+// QuarksLink represents the links to share/discover information between BOSH and Kube Native components
 type QuarksLink struct {
 	Type      string        `json:"type,omitempty"`
 	Address   string        `json:"address,omitempty"`

pkg/kube/controllers/boshdeployment/deployment_controller.go

@@ -174,7 +174,7 @@ func AddDeployment(ctx context.Context, config *config.Config, mgr manager.Manag
 	err = c.Watch(&source.Kind{Type: &corev1.Service{}}, &handler.EnqueueRequestsFromMapFunc{
 		ToRequests: handler.ToRequestsFunc(func(a handler.MapObject) []reconcile.Request {
 			// Get one request from one service at most
-			reconciles := make([]reconcile.Request, 0, 1)
+			reconciles := make([]reconcile.Request, 1)
 
 			svc := a.Object.(*corev1.Service)
 			if provider, ok := svc.GetAnnotations()[bdv1.AnnotationLinkProviderName]; ok {

pkg/kube/controllers/boshdeployment/deployment_reconciler.go

@@ -296,19 +296,21 @@ func (r *ReconcileBOSHDeployment) listLinkInfos(instance *bdv1.BOSHDeployment, m
 
 		serviceRecords, err := r.getServiceRecords(instance.Namespace, services.Items)
 		if err != nil {
-			return linkInfos, errors.New(fmt.Sprintf("Failed to get link services for: %s", instance.Name))
+			return linkInfos, errors.Wrap(err, fmt.Sprintf("Failed to get link services for: %s", instance.Name))
 		}
 
 		for qName := range quarksLinks {
 			if svcRecord, ok := serviceRecords[qName]; ok {
 				pods, err := r.listPodsFromSelector(instance.Namespace, svcRecord.selector)
 				if err != nil {
-					return linkInfos, errors.New(fmt.Sprintf("Failed to get link pods for: %s", instance.Name))
+					return linkInfos, errors.Wrap(err, fmt.Sprintf("Failed to get link pods for: %s", instance.Name))
 				}
 
 				var jobsInstances []bdm.JobInstance
 				for i, p := range pods {
-
+					if len(p.Status.PodIP) == 0 {
+						return linkInfos, fmt.Errorf("empty ip of kube native component: '%s/%s'", p.Namespace, p.Name)
+					}
 					jobsInstances = append(jobsInstances, bdm.JobInstance{
 						Name:      qName,
 						ID:        string(p.GetUID()),
@@ -340,6 +342,9 @@ func (r *ReconcileBOSHDeployment) listLinkInfos(instance *bdv1.BOSHDeployment, m
 	}
 
 	if len(quarksLinks) != 0 {
+		if manifest.Properties == nil {
+			manifest.Properties = map[string]interface{}{}
+		}
 		manifest.Properties["quarks_links"] = quarksLinks
 	}
 
@@ -358,7 +363,7 @@ func (r *ReconcileBOSHDeployment) getServiceRecords(namespace string, svcs []cor
 
 			svcRecords[providerName] = serviceRecord{
 				selector:  svc.Spec.Selector,
-				dnsRecord: fmt.Sprintf("%s.%s.svc.%s", svc.Name, namespace, bdm.GetClusterDomain()),
+				dnsRecord: fmt.Sprintf("%s.%s.svc.cluster.local", svc.Name, namespace),
 			}
 		}
 	}

testing/assets/gatherManifest.yml

@@ -325,7 +325,7 @@ instance_groups:
                 properties:
                   doppler:
                     grpc_port: 7765
-                  fooprop: 10001
+                    fooprop: 10001
             instances: null
             release: loggregator
             envs:

testing/assets/jobs-src/loggregator/doppler/job.MF

@@ -19,7 +19,7 @@ provides:
   type: doppler
   properties:
   - doppler.grpc_port
-  - fooprop
+  - doppler.fooprop
 - name: loggregator
   type: loggregator
   properties:
@@ -49,7 +49,7 @@ properties:
     description: Port for outgoing log messages via GRPC
     default: 8082
 
-  fooprop:
+  doppler.fooprop:
     description: Port for outgoing log messages via GRPC
     default: 10001
 

testing/assets/jobs-src/loggregator/loggregator_trafficcontroller/templates/bpm.yml.erb

@@ -22,7 +22,7 @@ processes:
       TRAFFIC_CONTROLLER_DISABLE_ACCESS_CONTROL: "<%= p("traffic_controller.disable_access_control") %>"
 
       FOOBARWITHLINKADDRESS: <%= link('doppler').address %>
-      FOOBARWITHLINKVALUES: <%= link('doppler').p("fooprop") %>
+      FOOBARWITHLINKVALUES: <%= link('doppler').p("doppler.fooprop") %>
       FOOBARWITHLINKNESTEDVALUES: <%= link('doppler').p("doppler.grpc_port") %>
       FOOBARWITHLINKINSTANCESINDEX: <%= link('doppler').instances[0].index %>
       FOOBARWITHLINKINSTANCESAZ: <%= link('doppler').instances[0].az %>

testing/boshmanifest/manifests.go

@@ -110,6 +110,28 @@ instance_groups:
           internal: 4223
 `
 
+// NatsSmokeTestWithExternalLinks has explicit BOSH links.
+// It can be used in integration tests.
+const NatsSmokeTestWithExternalLinks = `---
+name: test
+releases:
+- name: nats
+  version: "26"
+  url: docker.io/cfcontainerization
+  stemcell:
+    os: opensuse-42.3
+    version: 30.g9c91e77-30.80-7.0.0_257.gb97ced55
+instance_groups:
+- name: nats-smoke-tests
+  instances: 1
+  lifecycle: auto-errand
+  jobs:
+  - name: smoke-tests
+    release: nats
+    consumes:
+      nats: {from: nats}
+`
+
 // Drains is a small manifest with jobs that include drain scripts
 // It can be used in integration tests.
 const Drains = `---